agent: add reopenStore function for postgres; run notification tests with postgres (#1435)

src/Simplex/Messaging/Agent/Store.hs

@@ -73,6 +73,9 @@ createStore dbFilePath dbKey keepKey = Store.createDBStore dbFilePath dbKey keep
 closeStore :: DBStore -> IO ()
 closeStore = Store.closeDBStore
 
+reopenStore :: DBStore -> IO ()
+reopenStore = Store.reopenDBStore
+
 execSQL :: DB.Connection -> Text -> IO [Text]
 execSQL = Store.execSQL
 

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE QuasiQuotes #-}
@@ -6,7 +7,8 @@
 module Simplex.Messaging.Agent.Store.Postgres
   ( createDBStore,
     closeDBStore,
-    execSQL
+    reopenDBStore,
+    execSQL,
   )
 where
 
@@ -15,7 +17,7 @@ import Control.Monad (unless, void)
 import Data.Functor (($>))
 import Data.String (fromString)
 import Data.Text (Text)
-import Database.PostgreSQL.Simple (ConnectInfo (..), Only (..), defaultConnectInfo)
+import Database.PostgreSQL.Simple (ConnectInfo (..), Only (..))
 import qualified Database.PostgreSQL.Simple as PSQL
 import Database.PostgreSQL.Simple.SqlQQ (sql)
 import Simplex.Messaging.Agent.Store.Migrations (migrateSchema)
@@ -24,7 +26,7 @@ import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
 import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists)
 import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
 import Simplex.Messaging.Util (ifM)
-import UnliftIO.Exception (onException)
+import UnliftIO.Exception (bracketOnError, onException)
 import UnliftIO.MVar
 import UnliftIO.STM
 
@@ -44,11 +46,11 @@ createDBStore connectInfo schema migrations confirmMigrations = do
     Left e -> closeDBStore st $> Left e
 
 connectPostgresStore :: ConnectInfo -> String -> IO DBStore
-connectPostgresStore dbConnectInfo schema = do
-  (dbConn, dbNew) <- connectDB dbConnectInfo schema -- TODO [postgres] analogue for dbBusyLoop?
+connectPostgresStore dbConnectInfo dbSchema = do
+  (dbConn, dbNew) <- connectDB dbConnectInfo dbSchema -- TODO [postgres] analogue for dbBusyLoop?
   dbConnection <- newMVar dbConn
   dbClosed <- newTVarIO False
-  pure DBStore {dbConnectInfo, dbConnection, dbNew, dbClosed}
+  pure DBStore {dbConnectInfo, dbSchema, dbConnection, dbNew, dbClosed}
 
 connectDB :: ConnectInfo -> String -> IO (DB.Connection, Bool)
 connectDB dbConnectInfo schema = do
@@ -81,6 +83,22 @@ closeDBStore st@DBStore {dbClosed} =
       DB.close conn
       atomically $ writeTVar dbClosed True
 
+openPostgresStore_ :: DBStore -> IO ()
+openPostgresStore_ DBStore {dbConnectInfo, dbSchema, dbConnection, dbClosed} =
+  bracketOnError
+    (takeMVar dbConnection)
+    (tryPutMVar dbConnection)
+    $ \_dbConn -> do
+      (dbConn, _dbNew) <- connectDB dbConnectInfo dbSchema
+      atomically $ writeTVar dbClosed False
+      putMVar dbConnection dbConn
+
+reopenDBStore :: DBStore -> IO ()
+reopenDBStore st@DBStore {dbClosed} =
+  ifM (readTVarIO dbClosed) open (putStrLn "reopenDBStore: already opened")
+  where
+    open = openPostgresStore_ st
+
 -- TODO [postgres] not necessary for postgres (used for ExecAgentStoreSQL, ExecChatStoreSQL)
 execSQL :: PSQL.Connection -> Text -> IO [Text]
 execSQL _db _query = throwIO (userError "not implemented")

src/Simplex/Messaging/Agent/Store/Postgres/Common.hs

@@ -17,6 +17,7 @@ import UnliftIO.STM
 -- TODO [postgres] use log_min_duration_statement instead of custom slow queries (SQLite's Connection type)
 data DBStore = DBStore
   { dbConnectInfo :: PSQL.ConnectInfo,
+    dbSchema :: String,
     dbConnection :: MVar PSQL.Connection,
     dbClosed :: TVar Bool,
     dbNew :: Bool

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20241210_initial.hs

@@ -208,7 +208,7 @@ CREATE TABLE ntf_tokens(
   tkn_action BYTEA,
   created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
   updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
-  ntf_mode TEXT NULL,
+  ntf_mode BYTEA NULL,
   PRIMARY KEY(provider, device_token, ntf_host, ntf_port),
   FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
   ON DELETE RESTRICT ON UPDATE CASCADE
@@ -222,8 +222,8 @@ CREATE TABLE ntf_subscriptions(
   ntf_port TEXT NOT NULL,
   ntf_sub_id BYTEA,
   ntf_sub_status TEXT NOT NULL,
-  ntf_sub_action TEXT,
-  ntf_sub_smp_action TEXT,
+  ntf_sub_action BYTEA,
+  ntf_sub_smp_action BYTEA,
   ntf_sub_action_ts TIMESTAMPTZ,
   updated_by_supervisor SMALLINT NOT NULL DEFAULT 0,
   created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),

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

@@ -27,13 +27,12 @@
 module Simplex.Messaging.Agent.Store.SQLite
   ( createDBStore,
     closeDBStore,
+    reopenDBStore,
     execSQL,
     -- used in Simplex.Chat.Archive
     sqlString,
     keyString,
     storeKey,
-    -- used in Simplex.Chat.Mobile and tests
-    reopenSQLiteStore,
     -- used in tests
     connectSQLiteStore,
     openSQLiteStore,
@@ -127,14 +126,14 @@ openSQLiteStore_ DBStore {dbConnection, dbFilePath, dbKey, dbClosed} key keepKey
         writeTVar dbKey $! storeKey key keepKey
       putMVar dbConnection DB.Connection {conn, slow}
 
-reopenSQLiteStore :: DBStore -> IO ()
-reopenSQLiteStore st@DBStore {dbKey, dbClosed} =
-  ifM (readTVarIO dbClosed) open (putStrLn "reopenSQLiteStore: already opened")
+reopenDBStore :: DBStore -> IO ()
+reopenDBStore st@DBStore {dbKey, dbClosed} =
+  ifM (readTVarIO dbClosed) open (putStrLn "reopenDBStore: already opened")
   where
     open =
       readTVarIO dbKey >>= \case
         Just key -> openSQLiteStore_ st key True
-        Nothing -> fail "reopenSQLiteStore: no key"
+        Nothing -> fail "reopenDBStore: no key"
 
 keyString :: ScrubbedBytes -> Text
 keyString = sqlString . safeDecodeUtf8 . BA.convert