smp server: store service with its first queue

src/Simplex/Messaging/Server.hs

@@ -745,12 +745,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
         _ -> pure ()
 
     getClientService :: s -> TVar ChaChaDRG -> Int -> SMPServiceRole -> X.CertificateChain -> XV.Fingerprint -> ExceptT TransportError IO ServiceId
-    getClientService ms g idSize role cert fp = do
+    getClientService ms g idSize role _ (XV.Fingerprint fp) = do
       newServiceId <- EntityId <$> atomically (C.randomBytes idSize g)
-      ts <- liftIO getSystemDate
-      let sr = ServiceRec {serviceId = newServiceId, serviceRole = role, serviceCert = cert, serviceCertHash = fp, serviceCreatedAt = ts}
       withExceptT (const $ TEHandshake BAD_SERVICE) $ ExceptT $
-        getCreateService @(StoreQueue s) (queueStore ms) sr
+        fromMaybe newServiceId <$$> getServiceId @(StoreQueue s) (queueStore ms) role fp
 
     controlPortThread_ :: ServerConfig s -> [M s ()]
     controlPortThread_ ServerConfig {controlPort = Just port} = [runCPServer port]
@@ -1076,7 +1074,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do
         expCfg <- asks $ inactiveClientExpiration . config
         th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
         labelMyThread . B.unpack $ "client $" <> encode sessionId
-        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms c, receive h ms c] <> disconnectThread_ c s expCfg
+        raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms (getPeerCertChain $ connection h) c, receive h ms c] <> disconnectThread_ c s expCfg
     disconnectThread_ :: Client s -> Server s -> Maybe ExpirationConfig -> [M s ()]
     disconnectThread_ c s (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c s)]
     disconnectThread_ _ _ _ = []
@@ -1405,15 +1403,16 @@ forkClient Client {endThreads, endThreadSeq} label action = do
     action `finally` atomically (modifyTVar' endThreads $ IM.delete tId)
   mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId
 
-client :: forall s. MsgStoreClass s => Server s -> s -> Client s -> M s ()
+client :: forall s. MsgStoreClass s => Server s -> s -> X.CertificateChain -> Client s -> M s ()
 client
   Server {subscribers, ntfSubscribers}
   ms
+  serviceCert
   clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do
     labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
     stats <- asks serverStats
     authDelay <- asks $ authErrorDelay . config
-    let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams')
+    let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> clntService
         process batchSubs t acc@(rs, msgs) =
           (maybe acc (\(!r, !msg_) -> (r : rs, maybe msgs (: msgs) msg_)))
             <$> processCommand clntServiceId batchSubs t
@@ -1424,6 +1423,7 @@ client
       foldrM (process batchSubs) ([], []) batch
         >>= \(rs_, msgs) -> mapM_ (liftIO . sendResponses clnt stats authDelay processedAt . (,msgs)) (L.nonEmpty rs_)
   where
+    clntService = peerClientService =<< thAuth thParams'
     prepareBatchSubs ::
       Maybe ServiceId ->
       NonEmpty (VerifiedTransmission s) ->
@@ -1445,7 +1445,13 @@ client
       where
         ifNotNull qs f = if null qs then pure M.empty else f qs
         setService :: (PartyI p, ServiceParty p) => SParty p -> Maybe ServiceId -> [StoreQueue s] -> ExceptT ErrorType IO (Map RecipientId (Either ErrorType ()))
-        setService party sId = ExceptT . setQueueServices (queueStore ms) party sId
+        setService party sId qs = forM_ clntService storeService >> ExceptT (setQueueServices (queueStore ms) party sId qs)
+
+    storeService :: THPeerClientService -> ExceptT ErrorType IO ()
+    storeService THClientService {serviceId, serviceRole, serviceCertHash} = do
+      serviceCreatedAt <- liftIO getSystemDate
+      serviceId' <- ExceptT $ getCreateService @(StoreQueue s) (queueStore ms) ServiceRec {serviceId, serviceRole, serviceCert, serviceCertHash, serviceCreatedAt}
+      unless (serviceId' == serviceId) $ throwE SERVICE
 
     processProxiedCmd :: Transmission (Command 'ProxiedClient) -> M s (Maybe ResponseAndMessage)
     processProxiedCmd (corrId, EntityId sessId, command) = (\t -> ((corrId, EntityId sessId, t), Nothing)) <$$> case command of
@@ -1673,7 +1679,7 @@ client
                                   updatedAt,
                                   rcvServiceId = clntServiceId
                                 }
-                        liftIO (addQueue ms rcvId qr) >>= \case
+                        liftIO (runExceptT (forM_ clntService storeService) $>> addQueue ms rcvId qr) >>= \case
                           Left DUPLICATE_ -- TODO [short links] possibly, we somehow need to understand which IDs caused collision to retry if it's not client-supplied?
                             | clntIds -> pure $ ERR AUTH -- no retry on collision if sender ID is client-supplied
                             | otherwise -> tryCreate (n - 1)

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

@@ -339,6 +339,15 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
       rId = recipientId sq
       qr = queueRec sq
 
+  getServiceId :: PostgresQueueStore q -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId))
+  getServiceId st serviceRole fp =
+    E.uninterruptibleMask_ $ runExceptT $ withDB "getServiceId" st $ \db ->
+      traverse checkRole <$> maybeFirstRow id (DB.query db "SELECT service_id, service_role FROM services WHERE service_cert_hash = ?" (Only (Binary fp)))
+    where
+      checkRole (serviceId, role)
+        | role == serviceRole = Right serviceId
+        | otherwise = Left SERVICE
+
   getCreateService :: PostgresQueueStore q -> ServiceRec -> IO (Either ErrorType ServiceId)
   getCreateService st sr@ServiceRec {serviceId = newSrvId, serviceRole, serviceCertHash = XV.Fingerprint fp} =
     withLockMap (serviceLocks st) fp "getCreateService" $ E.uninterruptibleMask_ $ runExceptT $ do

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

@@ -286,6 +286,14 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
         mapM_ (removeNotifier st) $ notifier q
         pure q
 
+  getServiceId :: STMQueueStore q -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId))
+  getServiceId st serviceRole fp =
+    traverse checkRole <$> (TM.lookupIO fp (serviceCerts st) $>>= (`TM.lookupIO` services st))
+    where
+      checkRole STMService {serviceRec = ServiceRec {serviceId, serviceRole = role}}
+        | role == serviceRole = Right serviceId
+        | otherwise = Left SERVICE
+
   getCreateService :: STMQueueStore q -> ServiceRec -> IO (Either ErrorType ServiceId)
   getCreateService st sr@ServiceRec {serviceId = newSrvId, serviceRole, serviceCertHash = XV.Fingerprint fp} =
     TM.lookupIO fp serviceCerts
@@ -363,7 +371,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
   getServiceQueueCountHash :: (PartyI p, ServiceParty p) => STMQueueStore q -> SParty p -> ServiceId -> IO (Either ErrorType (Int64, IdsHash))
   getServiceQueueCountHash st party serviceId =
     TM.lookupIO serviceId (services st) >>=
-      maybe (pure $ Left AUTH) (fmap (Right . first (fromIntegral . S.size)) . readTVarIO . serviceSel)
+      maybe (pure $ Right (0, mempty)) (fmap (Right . first (fromIntegral . S.size)) . readTVarIO . serviceSel)
     where
       serviceSel :: STMService -> TVar (Set QueueId, IdsHash)
       serviceSel = case party of

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

@@ -17,6 +17,7 @@ import Data.Map.Strict (Map)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.SystemTime
+import Simplex.Messaging.Transport (SMPServiceRole)
 
 class StoreQueueClass q where
   recipientId :: q -> RecipientId
@@ -43,6 +44,7 @@ class StoreQueueClass q => QueueStoreClass q s where
   unblockQueue :: s -> q -> IO (Either ErrorType ())
   updateQueueTime :: s -> q -> SystemDate -> IO (Either ErrorType QueueRec)
   deleteStoreQueue :: s -> q -> IO (Either ErrorType QueueRec)
+  getServiceId :: s -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId))
   getCreateService :: s -> ServiceRec -> IO (Either ErrorType ServiceId)
   setQueueService :: (PartyI p, ServiceParty p) => s -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ())
   setQueueServices :: (PartyI p, ServiceParty p) => s -> SParty p -> Maybe ServiceId -> [q] -> IO (Either ErrorType (Map RecipientId (Either ErrorType ())))

tests/AgentTests/FunctionalAPITests.hs

@@ -4719,14 +4719,10 @@ testServiceIdChangeOnReconnect ps@(_, ASType qs _) = do
       ("", "", SERVICE_UP _ _) <- nGet service
       ("", "", UP _ [_]) <- nGet user
       pure ()
-    -- Phase 3: Normal reconnect - service should subscribe normally
     ("", "", SERVICE_DOWN _ _) <- nGet service
     ("", "", DOWN _ [_]) <- nGet user
     withSmpServerStoreLogOn ps testPort $ \_ -> do
-      liftIO $ getInAnyOrder service
-        [ \case ("", "", AEvt SAENone (SERVICE_UP _ _)) -> True; _ -> False,
-          \case ("", "", AEvt SAENone (SERVICE_ALL _)) -> True; _ -> False
-        ]
+      ("", "", SERVICE_UP _ (SMP.ServiceSubResult (Just SMP.SSErrorServiceId {}) (SMP.ServiceSub _ 0 _))) <- nGet service
       ("", "", UP _ [_]) <- nGet user
       pure ()
 

tests/ServerTests.hs

@@ -53,6 +53,7 @@ import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..
 import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..))
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Credentials
+import Simplex.Messaging.Transport.Server (loadServerCredential)
 import System.Directory (doesFileExist, removeFile)
 import System.IO (IOMode (..), withFile)
 import System.Timeout
@@ -91,6 +92,7 @@ serverTests = do
     testServiceDeliverSubscribe
     testServiceUpgradeAndDowngrade
     testServiceSubsTotalCount
+    testServiceStoredWithQueue
   describe "Store log" testWithStoreLog
   describe "Restore messages" testRestoreMessages
   describe "Restore messages (old / v2)" testRestoreExpireMessages
@@ -935,9 +937,35 @@ testServiceSubsTotalCount =
         threadDelay 1500000
         B.readFile testPrometheusMetricsFile >>= \m -> readServiceSubsMetric (B.unpack m) `shouldBe` Just 3
 
+testServiceStoredWithQueue :: SpecWith (ASrvTransport, AStoreType)
+testServiceStoredWithQueue =
+  it "should store service with its first queue" $ \(at@(ATransport t), msType) -> do
+    g <- C.newRandom
+    tlsCred <- loadServerCredential ntfTestServerCredentials
+    serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
+    (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+    (dhPub, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
+    let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {prometheusInterval = Just 1}
+        servicesCount = do
+          threadDelay 1500000
+          readMetric "simplex_smp_rcv_services_count " . B.unpack <$> B.readFile testPrometheusMetricsFile
+    withSmpServerConfigOn at cfg' testPort $ \_ -> runSMPClient t $ \h -> do
+      Resp "1" NoEntity (Ids rId _ _) <- signSendRecv h rKey ("1", NoEntity, New rPub dhPub)
+      runSMPServiceClient t (tlsCred, serviceKeys) $ \sh -> do
+        servicesCount `shouldReturn` Just 0
+        runSMPServiceClient t (tlsCred, serviceKeys) $ \sh' -> do
+          Resp "2" _ (SOK (Just _)) <- serviceSignSendRecv sh' rKey servicePK ("2", rId, SUB)
+          servicesCount `shouldReturn` Just 1
+        Resp "3" _ (ERR SERVICE) <- serviceSignSendRecv sh rKey servicePK ("3", rId, SUB)
+        pure ()
+      testNtfServiceClient t serviceKeys (\_ -> pure ()) `shouldThrow` anyErrorCall
+
 readServiceSubsMetric :: String -> Maybe Int
-readServiceSubsMetric content =
-  case filter ("simplex_smp_subscribtion_service_subs_total " `isPrefixOf`) (lines content) of
+readServiceSubsMetric = readMetric "simplex_smp_subscribtion_service_subs_total "
+
+readMetric :: String -> String -> Maybe Int
+readMetric metric content =
+  case filter (metric `isPrefixOf`) (lines content) of
     (line : _) -> case words line of
       [_, val, _] -> readMaybe val
       [_, val] -> readMaybe val