ntf server: PostgreSQL database storage (#1519)

* ntf server: PostgreSQL database storage

* ntf server: import/export stubs

* ntf server postgres db functions

* some notifications tests pass

* notifications tests pass

* import/export notification store logs

* fix ntf server CLI

* log in parralel

* update subscription statuses using executeMany

* fix import/export

* refactor

* fix queries

* prohibit token_id and subscription_id updates, dont update tokens and subscriptions on conflict, improve server insertion, remove duplicate tokens for import, remove subscriptions without tokens for import

* comment

* remame

* increase test delay

Author: epoberezkin

simplexmq.cabal

@@ -216,15 +216,6 @@ library
           Simplex.FileTransfer.Server.Stats
           Simplex.FileTransfer.Server.Store
           Simplex.FileTransfer.Server.StoreLog
-          Simplex.Messaging.Notifications.Server
-          Simplex.Messaging.Notifications.Server.Control
-          Simplex.Messaging.Notifications.Server.Env
-          Simplex.Messaging.Notifications.Server.Main
-          Simplex.Messaging.Notifications.Server.Push.APNS
-          Simplex.Messaging.Notifications.Server.Push.APNS.Internal
-          Simplex.Messaging.Notifications.Server.Stats
-          Simplex.Messaging.Notifications.Server.Store
-          Simplex.Messaging.Notifications.Server.StoreLog
           Simplex.Messaging.Server
           Simplex.Messaging.Server.CLI
           Simplex.Messaging.Server.Control
@@ -257,6 +248,18 @@ library
 
       if flag(server_postgres)
           exposed-modules:
+              Simplex.Messaging.Notifications.Server
+              Simplex.Messaging.Notifications.Server.Control
+              Simplex.Messaging.Notifications.Server.Env
+              Simplex.Messaging.Notifications.Server.Main
+              Simplex.Messaging.Notifications.Server.Push.APNS
+              Simplex.Messaging.Notifications.Server.Push.APNS.Internal
+              Simplex.Messaging.Notifications.Server.Stats
+              Simplex.Messaging.Notifications.Server.Store
+              Simplex.Messaging.Notifications.Server.Store.Migrations
+              Simplex.Messaging.Notifications.Server.Store.Postgres
+              Simplex.Messaging.Notifications.Server.Store.Types
+              Simplex.Messaging.Notifications.Server.StoreLog
               Simplex.Messaging.Server.QueueStore.Postgres
               Simplex.Messaging.Server.QueueStore.Postgres.Migrations
   other-modules:
@@ -340,6 +343,8 @@ library
       , sqlcipher-simple ==0.4.*
   if flag(server_postgres)
     cpp-options: -DdbServerPostgres
+    build-depends:
+        hex-text ==0.1.*
   if impl(ghc >= 9.6.2)
     build-depends:
         bytestring ==0.11.*
@@ -352,6 +357,10 @@ library
 executable ntf-server
   if flag(client_library)
       buildable: False
+  if flag(server_postgres)
+    cpp-options: -DdbServerPostgres
+  else
+    buildable: False
   main-is: Main.hs
   other-modules:
       Paths_simplexmq
@@ -444,7 +453,6 @@ test-suite simplexmq-test
       AgentTests.EqInstances
       AgentTests.FunctionalAPITests
       AgentTests.MigrationTests
-      AgentTests.NotificationTests
       AgentTests.ServerChoice
       AgentTests.ShortLinkTests
       CLITests
@@ -460,8 +468,6 @@ test-suite simplexmq-test
       CoreTests.UtilTests
       CoreTests.VersionRangeTests
       FileDescriptionTests
-      NtfClient
-      NtfServerTests
       RemoteControl
       ServerTests
       SMPAgentClient
@@ -484,6 +490,9 @@ test-suite simplexmq-test
           AgentTests.SQLiteTests
   if flag(server_postgres)
       other-modules:
+          AgentTests.NotificationTests
+          NtfClient
+          NtfServerTests
           ServerTests.SchemaDump  
   hs-source-dirs:
       tests
@@ -537,6 +546,8 @@ test-suite simplexmq-test
     , warp-tls
     , yaml
   default-language: Haskell2010
+  if flag(server_postgres)
+    cpp-options: -DdbServerPostgres
   if flag(client_postgres)
     cpp-options: -DdbPostgres
   else
@@ -550,5 +561,3 @@ test-suite simplexmq-test
   if flag(client_postgres) || flag(server_postgres)
     build-depends:
         postgresql-simple ==0.7.*
-  if flag(server_postgres)
-    cpp-options: -DdbServerPostgres

src/Simplex/Messaging/Client/Agent.hs

@@ -412,14 +412,22 @@ removeSubscription :: SMPClientAgent -> SMPServer -> SMPSub -> STM ()
 removeSubscription = removeSub_ . srvSubs
 {-# INLINE removeSubscription #-}
 
+removePendingSub :: SMPClientAgent -> SMPServer -> SMPSub -> STM ()
+removePendingSub = removeSub_ . pendingSrvSubs
+{-# INLINE removePendingSub #-}
+
 removeSub_ :: TMap SMPServer (TMap SMPSub s) -> SMPServer -> SMPSub -> STM ()
 removeSub_ subs srv s = TM.lookup srv subs >>= mapM_ (TM.delete s)
 
+removeSubscriptions :: SMPClientAgent -> SMPServer -> SMPSubParty -> [QueueId] -> STM ()
+removeSubscriptions = removeSubs_ . srvSubs
+{-# INLINE removeSubscriptions #-}
+
 removePendingSubs :: SMPClientAgent -> SMPServer -> SMPSubParty -> [QueueId] -> STM ()
 removePendingSubs = removeSubs_ . pendingSrvSubs
 {-# INLINE removePendingSubs #-}
 
-removeSubs_ :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey) -> SMPServer -> SMPSubParty -> [QueueId] -> STM ()
+removeSubs_ :: TMap SMPServer (TMap SMPSub s) -> SMPServer -> SMPSubParty -> [QueueId] -> STM ()
 removeSubs_ subs srv party qs = TM.lookup srv subs >>= mapM_ (`modifyTVar'` (`M.withoutKeys` ss))
   where
     ss = S.fromList $ map (party,) qs

src/Simplex/Messaging/Encoding.hs

@@ -143,7 +143,7 @@ instance Encoding Large where
 instance Encoding SystemTime where
   smpEncode = smpEncode . systemSeconds
   {-# INLINE smpEncode #-}
-  smpP = MkSystemTime <$> smpP <*> pure 0
+  smpP = (`MkSystemTime` 0) <$> smpP
   {-# INLINE smpP #-}
 
 _smpP :: Encoding a => Parser a

src/Simplex/Messaging/Encoding/String.hs

@@ -140,7 +140,7 @@ instance StrEncoding Int64 where
 
 instance StrEncoding SystemTime where
   strEncode = strEncode . systemSeconds
-  strP = MkSystemTime <$> strP <*> pure 0
+  strP = (`MkSystemTime` 0) <$> strP
 
 instance StrEncoding UTCTime where
   strEncode = B.pack . iso8601Show

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -517,8 +517,11 @@ instance Encoding NtfSubStatus where
 
 instance StrEncoding NtfSubStatus where
   strEncode = smpEncode
+  {-# INLINE strEncode #-}
   strP = smpP
+  {-# INLINE strP #-}
 
+-- TODO [ntfdb] check what happens in agent when token in not yet registered
 data NtfTknStatus
   = -- | Token created in DB
     NTNew
@@ -534,6 +537,17 @@ data NtfTknStatus
     NTExpired
   deriving (Eq, Show)
 
+allowNtfSubCommands :: NtfTknStatus -> Bool
+allowNtfSubCommands = \case
+  NTNew -> False
+  NTRegistered -> False
+  -- TODO [ntfdb] we could have separate statuses to show whether it became invalid
+  -- after verification (allow commands) or before (do not allow)
+  NTInvalid _ -> True
+  NTConfirmed -> False
+  NTActive -> True
+  NTExpired -> True
+
 instance Encoding NtfTknStatus where
   smpEncode = \case
     NTNew -> "NEW"