SMP commands for notifications (NKEY/NID and NSUB/NMSG) with separate queue ID and key (#199)

* SMP commands for notifications (LSTN, NTFY) with separate queue IDs and keys

* rename Notifier types

* remove notify key and id from NEW and IDS commands (TODO add other commands)

* fix StoreLog serialization

* add commands for managing notifications

* add notification subscribers to server state, add notifier ID and key to store log

* add notifier ID and key to the queue

* refactor END notification to work for both types of subscriptions, deliver message notification (NMSG)

* process NSUB command - subscribe to message notifications

* test for message notifications

* fix SMP client function for NSUB command

* fix parse/serialize NID command

* refactor use ifM

* check duplicate notifier ID only against other notifier IDs

* refactor getQueue

* test notifier ID and key with store log

* Update src/Simplex/Messaging/Client.hs

Co-authored-by: Efim Poberezkin <8711996+efim-poberezkin@users.noreply.github.com>

* Update src/Simplex/Messaging/Server.hs

Co-authored-by: Efim Poberezkin <8711996+efim-poberezkin@users.noreply.github.com>

* store log: s/NOTIFY/NOTIFIER/

Co-authored-by: Efim Poberezkin <8711996+efim-poberezkin@users.noreply.github.com>

src/Simplex/Messaging/Server/Env/STM.hs

@@ -46,11 +46,14 @@ data Env = Env
 
 data Server = Server
   { subscribedQ :: TBQueue (RecipientId, Client),
-    subscribers :: TVar (Map RecipientId Client)
+    subscribers :: TVar (Map RecipientId Client),
+    ntfSubscribedQ :: TBQueue (NotifierId, Client),
+    notifiers :: TVar (Map NotifierId Client)
   }
 
 data Client = Client
   { subscriptions :: TVar (Map RecipientId Sub),
+    ntfSubscriptions :: TVar (Map NotifierId ()),
     rcvQ :: TBQueue Transmission,
     sndQ :: TBQueue Transmission
   }
@@ -66,14 +69,17 @@ newServer :: Natural -> STM Server
 newServer qSize = do
   subscribedQ <- newTBQueue qSize
   subscribers <- newTVar M.empty
-  return Server {subscribedQ, subscribers}
+  ntfSubscribedQ <- newTBQueue qSize
+  notifiers <- newTVar M.empty
+  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers}
 
 newClient :: Natural -> STM Client
 newClient qSize = do
   subscriptions <- newTVar M.empty
+  ntfSubscriptions <- newTVar M.empty
   rcvQ <- newTBQueue qSize
   sndQ <- newTBQueue qSize
-  return Client {subscriptions, rcvQ, sndQ}
+  return Client {subscriptions, ntfSubscriptions, rcvQ, sndQ}
 
 newSubscription :: STM Sub
 newSubscription = do
@@ -94,7 +100,17 @@ newEnv config = do
     restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)
     restoreQueues queueStore s = do
       (queues, s') <- liftIO $ readWriteStoreLog s
-      atomically $ modifyTVar queueStore $ \d -> d {queues, senders = M.foldr' addSender M.empty queues}
+      atomically $
+        modifyTVar queueStore $ \d ->
+          d
+            { queues,
+              senders = M.foldr' addSender M.empty queues,
+              notifiers = M.foldr' addNotifier M.empty queues
+            }
       pure s'
     addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
     addSender q = M.insert (senderId q) (recipientId q)
+    addNotifier :: QueueRec -> Map NotifierId RecipientId -> Map NotifierId RecipientId
+    addNotifier q = case notifier q of
+      Nothing -> id
+      Just (nId, _) -> M.insert nId (recipientId q)

src/Simplex/Messaging/Server/QueueStore.hs

@@ -8,10 +8,11 @@ module Simplex.Messaging.Server.QueueStore where
 import Simplex.Messaging.Protocol
 
 data QueueRec = QueueRec
-  { recipientId :: QueueId,
-    senderId :: QueueId,
+  { recipientId :: RecipientId,
+    senderId :: SenderId,
     recipientKey :: RecipientPublicKey,
     senderKey :: Maybe SenderPublicKey,
+    notifier :: Maybe (NotifierId, NotifierPublicKey),
     status :: QueueStatus
   }
 
@@ -21,6 +22,7 @@ class MonadQueueStore s m where
   addQueue :: s -> RecipientPublicKey -> (RecipientId, SenderId) -> m (Either ErrorType ())
   getQueue :: s -> SParty (a :: Party) -> QueueId -> m (Either ErrorType QueueRec)
   secureQueue :: s -> RecipientId -> SenderPublicKey -> m (Either ErrorType ())
+  addQueueNotifier :: s -> RecipientId -> NotifierId -> NotifierPublicKey -> m (Either ErrorType ())
   suspendQueue :: s -> RecipientId -> m (Either ErrorType ())
   deleteQueue :: s -> RecipientId -> m (Either ErrorType ())
 
@@ -31,5 +33,6 @@ mkQueueRec recipientKey (recipientId, senderId) =
       senderId,
       recipientKey,
       senderKey = Nothing,
+      notifier = Nothing,
       status = QueueActive
     }

src/Simplex/Messaging/Server/QueueStore/STM.hs

@@ -19,13 +19,14 @@ import UnliftIO.STM
 
 data QueueStoreData = QueueStoreData
   { queues :: Map RecipientId QueueRec,
-    senders :: Map SenderId RecipientId
+    senders :: Map SenderId RecipientId,
+    notifiers :: Map NotifierId RecipientId
   }
 
 type QueueStore = TVar QueueStoreData
 
 newQueueStore :: STM QueueStore
-newQueueStore = newTVar QueueStoreData {queues = M.empty, senders = M.empty}
+newQueueStore = newTVar QueueStoreData {queues = M.empty, senders = M.empty, notifiers = M.empty}
 
 instance MonadQueueStore QueueStore STM where
   addQueue :: QueueStore -> RecipientPublicKey -> (RecipientId, SenderId) -> STM (Either ErrorType ())
@@ -42,22 +43,47 @@ instance MonadQueueStore QueueStore STM where
         return $ Right ()
 
   getQueue :: QueueStore -> SParty (p :: Party) -> QueueId -> STM (Either ErrorType QueueRec)
-  getQueue store SRecipient rId = do
-    cs <- readTVar store
-    return $ getRcpQueue cs rId
-  getQueue store SSender sId = do
-    cs <- readTVar store
-    let rId = M.lookup sId $ senders cs
-    return $ maybe (Left AUTH) (getRcpQueue cs) rId
-  getQueue _ SBroker _ =
-    return $ Left INTERNAL
+  getQueue st party qId = do
+    cs <- readTVar st
+    pure $ case party of
+      SRecipient -> getRcpQueue cs qId
+      SSender -> getPartyQueue cs senders
+      SNotifier -> getPartyQueue cs notifiers
+      SBroker -> Left INTERNAL
+    where
+      getPartyQueue ::
+        QueueStoreData ->
+        (QueueStoreData -> Map QueueId RecipientId) ->
+        Either ErrorType QueueRec
+      getPartyQueue cs recipientIds =
+        case M.lookup qId $ recipientIds cs of
+          Just rId -> getRcpQueue cs rId
+          Nothing -> Left AUTH
 
+  secureQueue :: QueueStore -> RecipientId -> SenderPublicKey -> STM (Either ErrorType ())
   secureQueue store rId sKey =
     updateQueues store rId $ \cs c ->
       case senderKey c of
         Just _ -> (Left AUTH, cs)
         _ -> (Right (), cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})
 
+  addQueueNotifier :: QueueStore -> RecipientId -> NotifierId -> NotifierPublicKey -> STM (Either ErrorType ())
+  addQueueNotifier store rId nId nKey = do
+    cs@QueueStoreData {queues, notifiers} <- readTVar store
+    if M.member nId notifiers
+      then pure $ Left DUPLICATE_
+      else case M.lookup rId queues of
+        Nothing -> pure $ Left AUTH
+        Just q -> case notifier q of
+          Just _ -> pure $ Left AUTH
+          _ -> do
+            writeTVar store $
+              cs
+                { queues = M.insert rId q {notifier = Just (nId, nKey)} queues,
+                  notifiers = M.insert nId rId notifiers
+                }
+            pure $ Right ()
+
   suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
   suspendQueue store rId =
     updateQueues store rId $ \cs c ->

src/Simplex/Messaging/Server/StoreLog.hs

@@ -14,6 +14,7 @@ module Simplex.Messaging.Server.StoreLog
     closeStoreLog,
     logCreateQueue,
     logSecureQueue,
+    logAddNotifier,
     logDeleteQueue,
     readWriteStoreLog,
   )
@@ -50,36 +51,44 @@ data StoreLog (a :: IOMode) where
 data StoreLogRecord
   = CreateQueue QueueRec
   | SecureQueue QueueId SenderPublicKey
+  | AddNotifier QueueId NotifierId NotifierPublicKey
   | DeleteQueue QueueId
 
 storeLogRecordP :: Parser StoreLogRecord
 storeLogRecordP =
   "CREATE " *> createQueueP
     <|> "SECURE " *> secureQueueP
+    <|> "NOTIFIER " *> addNotifierP
     <|> "DELETE " *> (DeleteQueue <$> base64P)
   where
     createQueueP = CreateQueue <$> queueRecP
     secureQueueP = SecureQueue <$> base64P <* A.space <*> C.pubKeyP
+    addNotifierP =
+      AddNotifier <$> base64P <* A.space <*> base64P <* A.space <*> C.pubKeyP
     queueRecP = do
       recipientId <- "rid=" *> base64P <* A.space
       senderId <- "sid=" *> base64P <* A.space
       recipientKey <- "rk=" *> C.pubKeyP <* A.space
       senderKey <- "sk=" *> optional C.pubKeyP
-      pure QueueRec {recipientId, senderId, recipientKey, senderKey, status = QueueActive}
+      notifier <- optional $ (,) <$> (" nid=" *> base64P) <*> (" nk=" *> C.pubKeyP)
+      pure QueueRec {recipientId, senderId, recipientKey, senderKey, notifier, status = QueueActive}
 
 serializeStoreLogRecord :: StoreLogRecord -> ByteString
 serializeStoreLogRecord = \case
   CreateQueue q -> "CREATE " <> serializeQueue q
   SecureQueue rId sKey -> "SECURE " <> encode rId <> " " <> C.serializePubKey sKey
+  AddNotifier rId nId nKey -> B.unwords ["NOTIFIER", encode rId, encode nId, C.serializePubKey nKey]
   DeleteQueue rId -> "DELETE " <> encode rId
   where
-    serializeQueue QueueRec {recipientId, senderId, recipientKey, senderKey} =
+    serializeQueue QueueRec {recipientId, senderId, recipientKey, senderKey, notifier} =
       B.unwords
         [ "rid=" <> encode recipientId,
           "sid=" <> encode senderId,
           "rk=" <> C.serializePubKey recipientKey,
           "sk=" <> maybe "" C.serializePubKey senderKey
         ]
+        <> maybe "" serializeNotifier notifier
+    serializeNotifier (nId, nKey) = " nid=" <> encode nId <> " nk=" <> C.serializePubKey nKey
 
 openWriteStoreLog :: FilePath -> IO (StoreLog 'WriteMode)
 openWriteStoreLog f = WriteStoreLog f <$> openFile f WriteMode
@@ -110,6 +119,9 @@ logCreateQueue s = writeStoreLogRecord s . CreateQueue
 logSecureQueue :: StoreLog 'WriteMode -> QueueId -> SenderPublicKey -> IO ()
 logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey
 
+logAddNotifier :: StoreLog 'WriteMode -> QueueId -> NotifierId -> NotifierPublicKey -> IO ()
+logAddNotifier s qId nId nKey = writeStoreLogRecord s $ AddNotifier qId nId nKey
+
 logDeleteQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteQueue s = writeStoreLogRecord s . DeleteQueue
 
@@ -141,6 +153,7 @@ readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLo
     procLogRecord m = \case
       CreateQueue q -> M.insert (recipientId q) q m
       SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
+      AddNotifier qId nId nKey -> M.adjust (\q -> q {notifier = Just (nId, nKey)}) qId m
       DeleteQueue qId -> M.delete qId m
     printError :: LogParsingError -> IO ()
     printError (e, s) = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s