ntf: test notification subscription (#389)

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -126,6 +126,8 @@ deriving instance Show (NewNtfEntity e)
 
 data ANewNtfEntity = forall e. NtfEntityI e => ANE (SNtfEntity e) (NewNtfEntity e)
 
+deriving instance Show ANewNtfEntity
+
 instance NtfEntityI e => Encoding (NewNtfEntity e) where
   smpEncode = \case
     NewNtfTkn tkn verifyKey dhPubKey -> smpEncode ('T', tkn, verifyKey, dhPubKey)

src/Simplex/Messaging/Notifications/Server.hs

@@ -108,12 +108,6 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
       (srv, _sessId, ntfId, msg) <- atomically $ readTBQueue msgQ
       case msg of
         SMP.NMSG -> do
-          -- check when the last NMSG was received from this queue
-          -- update timestamp
-          -- check what was the last hidden notification was sent (and whether to this queue)
-          -- decide whether it should be sent as hidden or visible
-          -- construct and possibly encrypt notification
-          -- send it
           NtfPushServer {pushQ} <- asks pushServer
           st <- asks store
           atomically $
@@ -149,6 +143,8 @@ ntfPush s@NtfPushServer {pushQ} = liftIO . forever . runExceptT $ do
       atomically $ modifyTVar tknStatus $ \status' -> if status' == NTActive then NTActive else NTConfirmed
     (NTActive, PNCheckMessages) -> do
       deliverNotification pp tkn ntf
+    (NTActive, PNMessage _ _) -> do
+      deliverNotification pp tkn ntf
     _ -> do
       logError "bad notification token status"
   where
@@ -194,6 +190,9 @@ send h@THandle {thVersion = v} NtfServerClient {sndQ, sessionId, activeAt} = for
   void . liftIO $ tPut h (Nothing, encodeTransmission v sessionId t)
   atomically . writeTVar activeAt =<< liftIO getSystemTime
 
+-- instance Show a => Show (TVar a) where
+--   show x = unsafePerformIO $ show <$> readTVarIO x
+
 data VerificationResult = VRVerified NtfRequest | VRFailed
 
 verifyNtfTransmission ::

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

@@ -23,7 +23,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Server.Store
-import Simplex.Messaging.Protocol (CorrId, Transmission, SMPServer)
+import Simplex.Messaging.Protocol (CorrId, SMPServer, Transmission)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM

src/Simplex/Messaging/Notifications/Server/Store.hs

@@ -6,7 +6,6 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Notifications.Server.Store where
 
@@ -14,7 +13,8 @@ import Control.Concurrent.STM
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.Map.Strict as M
-import Data.Set (Set, insert)
+import Data.Set (Set)
+import qualified Data.Set as S
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Protocol (NtfPrivateSignKey)
@@ -154,12 +154,18 @@ mkNtfSubData (NewNtfSub tokenId smpQueue notifierKey) = do
 
 addNtfSubscription :: NtfStore -> NtfSubscriptionId -> NtfSubData -> STM (Maybe ())
 addNtfSubscription st subId sub@NtfSubData {smpQueue, tokenId} =
-  TM.lookup tokenId (tokenSubscriptions st) >>= mapM insertSub
+  TM.lookup tokenId (tokenSubscriptions st) >>= maybe newTokenSub pure >>= insertSub
   where
+    newTokenSub = do
+      ts <- newTVar S.empty
+      TM.insert tokenId ts $ tokenSubscriptions st
+      pure ts
     insertSub ts = do
-      modifyTVar' ts $ insert subId
+      modifyTVar' ts $ S.insert subId
       TM.insert subId sub $ subscriptions st
       TM.insert smpQueue subId (subscriptionLookup st)
+      -- return Nothing if subscription existed before
+      pure $ Just ()
 
 -- getNtfRec :: NtfStore -> SNtfEntity e -> NtfEntityId -> STM (Maybe (NtfEntityRec e))
 -- getNtfRec st ent entId = case ent of

tests/AgentTests.hs

@@ -12,6 +12,7 @@ module AgentTests (agentTests) where
 import AgentTests.ConnectionRequestTests
 import AgentTests.DoubleRatchetTests (doubleRatchetTests)
 import AgentTests.FunctionalAPITests (functionalAPITests)
+import AgentTests.NotificationTests (notificationTests)
 import AgentTests.SQLiteTests (storeTests)
 import AgentTests.SchemaDump (schemaDumpTest)
 import Control.Concurrent
@@ -36,6 +37,7 @@ agentTests (ATransport t) = do
   describe "Connection request" connectionRequestTests
   describe "Double ratchet tests" doubleRatchetTests
   describe "Functional API" $ functionalAPITests (ATransport t)
+  describe "Notification tests" $ notificationTests (ATransport t)
   describe "SQLite store" storeTests
   describe "SQLite schema dump" schemaDumpTest
   describe "SMP agent protocol syntax" $ syntaxTests t

tests/NtfClient.hs

@@ -18,8 +18,9 @@ import Control.Monad
 import Control.Monad.Except (runExceptT)
 import Control.Monad.IO.Unlift
 import Crypto.Random
-import Data.Aeson (FromJSON (..), ToJSON (..))
+import Data.Aeson (FromJSON (..), ToJSON (..), (.:))
 import qualified Data.Aeson as J
+import qualified Data.Aeson.Types as JT
 import Data.ByteString.Builder (lazyByteString)
 import Data.ByteString.Char8 (ByteString)
 import Data.Text (Text)
@@ -42,6 +43,7 @@ import Simplex.Messaging.Transport.HTTP2 (http2TLSParams)
 import Simplex.Messaging.Transport.HTTP2.Client
 import Simplex.Messaging.Transport.HTTP2.Server
 import Simplex.Messaging.Transport.KeepAlive
+import Test.Hspec
 import UnliftIO.Async
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
@@ -134,6 +136,9 @@ ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
       (Nothing, _, (CorrId corrId, qId, Right cmd)) <- tGet h
       pure (Nothing, corrId, qId, cmd)
 
+ntfTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
+ntfTest _ test' = runNtfTest test' `shouldReturn` ()
+
 data APNSMockRequest = APNSMockRequest
   { notification :: APNSNotification,
     sendApnsResponse :: APNSMockResponse -> IO ()
@@ -163,9 +168,16 @@ withAPNSMockServer = E.bracket (getAPNSMockServer apnsMockServerConfig) closeAPN
 
 deriving instance Generic APNSAlertBody
 
-deriving instance FromJSON APNSAlertBody
+instance FromJSON APNSAlertBody where
+  parseJSON (J.Object v) = do
+    title <- v .: "title"
+    subtitle <- v .: "subtitle"
+    body <- v .: "body"
+    pure APNSAlertObject {title, subtitle, body}
+  parseJSON (J.String v) = pure $ APNSAlertText v
+  parseJSON invalid = JT.prependFailure "parsing Coord failed, " (JT.typeMismatch "Object" invalid)
 
-instance FromJSON APNSNotificationBody where parseJSON = J.genericParseJSON apnsJSONOptions
+instance FromJSON APNSNotificationBody where parseJSON = J.genericParseJSON apnsJSONOptions {J.rejectUnknownFields = True}
 
 deriving instance FromJSON APNSNotification
 
@@ -185,8 +197,11 @@ getAPNSMockServer config@HTTP2ServerConfig {qSize} = do
             APNSRespError status reason ->
               sendResponse . H.responseBuilder status [] . lazyByteString $ J.encode APNSErrorResponse {reason}
       case J.decodeStrict' reqBody of
-        Just notification -> atomically $ writeTBQueue apnsQ APNSMockRequest {notification, sendApnsResponse}
-        _ -> sendApnsResponse $ APNSRespError N.badRequest400 "bad_request_body"
+        Just notification ->
+          atomically $ writeTBQueue apnsQ APNSMockRequest {notification, sendApnsResponse}
+        _ -> do
+          putStrLn $ "runAPNSMockServer J.decodeStrict' error, reqBody: " <> show reqBody
+          sendApnsResponse $ APNSRespError N.badRequest400 "bad_request_body"
 
 closeAPNSMockServer :: APNSMockServer -> IO ()
 closeAPNSMockServer APNSMockServer {action, http2Server} = do

tests/NtfServerTests.hs

@@ -1,23 +1,46 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 
 module NtfServerTests where
 
+import Control.Concurrent (threadDelay)
+import Control.Monad.Except (runExceptT)
+import qualified Data.Aeson as J
+import qualified Data.Aeson.Types as JT
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
+import Data.Text.Encoding (encodeUtf8)
 import NtfClient
-import ServerTests (sampleDhPubKey, samplePubKey, sampleSig)
+import SMPClient as SMP
+import ServerTests
+  ( createAndSecureQueue,
+    sampleDhPubKey,
+    samplePubKey,
+    sampleSig,
+    signSendRecv,
+    (#==),
+    _SEND,
+    pattern Resp,
+  )
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
-import Simplex.Messaging.Protocol
+import Simplex.Messaging.Notifications.Server.Push.APNS
+import Simplex.Messaging.Protocol hiding (notification)
 import Simplex.Messaging.Transport
 import Test.Hspec
+import UnliftIO.STM
 
 ntfServerTests :: ATransport -> Spec
 ntfServerTests t = do
-  describe "notifications server protocol syntax" $ ntfSyntaxTests t
+  describe "Notifications server protocol syntax" $ ntfSyntaxTests t
+  describe "Managing notification subscriptions" $ testNotificationSubscription t
 
 ntfSyntaxTests :: ATransport -> Spec
 ntfSyntaxTests (ATransport t) = do
@@ -34,3 +57,71 @@ ntfSyntaxTests (ATransport t) = do
       (Maybe C.ASignature, ByteString, ByteString, BrokerMsg) ->
       Expectation
     command >#> response = withAPNSMockServer $ \_ -> ntfServerTest t command `shouldReturn` response
+
+pattern RespNtf :: CorrId -> QueueId -> NtfResponse -> SignedTransmission NtfResponse
+pattern RespNtf corrId queueId command <- (_, _, (corrId, queueId, Right command))
+
+sendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> (Maybe C.ASignature, ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission NtfResponse)
+sendRecvNtf h@THandle {thVersion, sessionId} (sgn, corrId, qId, cmd) = do
+  let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
+  Right () <- tPut h (sgn, t)
+  tGet h
+
+signSendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission NtfResponse)
+signSendRecvNtf h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
+  let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
+  Right sig <- runExceptT $ C.sign pk t
+  Right () <- tPut h (Just sig, t)
+  tGet h
+
+(.->) :: J.Value -> J.Key -> Either String ByteString
+v .-> key =
+  let J.Object o = v
+   in U.decodeLenient . encodeUtf8 <$> JT.parseEither (J..: key) o
+
+testNotificationSubscription :: ATransport -> Spec
+testNotificationSubscription (ATransport t) =
+  it "should create new notification subscription and notify when message is received" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    (nPub, nKey) <- C.generateSignatureKeyPair C.SEd25519
+    (tknPub, tknKey) <- C.generateSignatureKeyPair C.SEd25519
+    (dhPub, dhPriv :: C.PrivateKeyX25519) <- C.generateKeyPair'
+    let tkn = DeviceToken PPApns "abcd"
+    withAPNSMockServer $ \APNSMockServer {apnsQ} ->
+      smpTest2 t $ \rh sh ->
+        ntfTest t $ \nh -> do
+          -- create queue
+          (sId, rId, rKey, _dhShared) <- createAndSecureQueue rh sPub
+          -- register and verify token
+          RespNtf "1" "" (NRTknId tId ntfDh) <- signSendRecvNtf nh tknKey ("1", "", TNEW $ NewNtfTkn tkn tknPub dhPub)
+          APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse = send} <-
+            atomically $ readTBQueue apnsQ
+          send APNSRespOk
+          let dhSecret = C.dh' ntfDh dhPriv
+              Right verification = ntfData .-> "verification"
+              Right nonce = C.cbNonce <$> ntfData .-> "nonce"
+              Right code = NtfRegCode <$> C.cbDecrypt dhSecret nonce verification
+          RespNtf "2" _ NROk <- signSendRecvNtf nh tknKey ("2", tId, TVFY code)
+          RespNtf "2a" _ (NRTkn NTActive) <- signSendRecvNtf nh tknKey ("2a", tId, TCHK)
+          -- enable queue notifications
+          Resp "3" _ (NID nId) <- signSendRecv rh rKey ("3", rId, NKEY nPub)
+          let srv = SMPServer SMP.testHost SMP.testPort SMP.testKeyHash
+              q = SMPQueueNtf srv nId
+          RespNtf "4" _ (NRSubId _subId) <- signSendRecvNtf nh tknKey ("4", "", SNEW $ NewNtfSub tId q nKey)
+          -- send message
+          threadDelay 50000
+          Resp "5" _ OK <- signSendRecv sh sKey ("5", sId, _SEND "hello")
+          -- receive notification
+          APNSMockRequest {notification, sendApnsResponse = send'} <- atomically $ readTBQueue apnsQ
+          let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData'} = notification
+              Right checkMessage = ntfData' .-> "checkMessage"
+              Right nonce' = C.cbNonce <$> ntfData' .-> "nonce"
+              Right smpQueueURI = C.cbDecrypt dhSecret nonce' checkMessage
+          smpQueueURI `shouldBe` strEncode srv <> "/" <> strEncode nId
+          send' APNSRespOk
+          -- receive message
+          let dec _nonce = C.cbDecrypt _dhShared (C.cbNonce _nonce)
+          Resp "" _ (MSG mId1 _ _ msg1) <- tGet rh
+          (dec mId1 msg1, Right "hello") #== "delivered from queue"
+          Resp "6" _ OK <- signSendRecv rh rKey ("6", rId, ACK mId1)
+          pure ()

tests/Test.hs

@@ -1,7 +1,6 @@
 {-# LANGUAGE TypeApplications #-}
 
 import AgentTests (agentTests)
-import AgentTests.NotificationTests (notificationTests)
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
 import CoreTests.VersionRangeTests
@@ -25,8 +24,6 @@ main = do
       describe "Version range" versionRangeTests
     describe "SMP server via TLS" $ serverTests (transport @TLS)
     describe "SMP server via WebSockets" $ serverTests (transport @WS)
-    describe "Notifications server" $ do
-      ntfServerTests (transport @TLS)
-      notificationTests (transport @TLS)
+    describe "Notifications server" $ ntfServerTests (transport @TLS)
     describe "SMP client agent" $ agentTests (transport @TLS)
   removeDirectoryRecursive "tests/tmp"