mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-03-30 16:26:02 +00:00
This reverts commit 1afcefa5e7.
This commit is contained in:
committed by
GitHub
parent
fdbfe0e8d1
commit
82aec2cd8f
@@ -211,6 +211,7 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
)
|
||||
where
|
||||
|
||||
import Control.Concurrent (threadDelay)
|
||||
import Control.Concurrent.STM (stateTVar)
|
||||
import Control.Monad.Except
|
||||
import Crypto.Random (ChaChaDRG, randomBytesGenerate)
|
||||
@@ -233,7 +234,7 @@ import Data.Ord (Down (..))
|
||||
import Data.Text (Text)
|
||||
import qualified Data.Text as T
|
||||
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
|
||||
import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
|
||||
import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
|
||||
import Data.Word (Word32)
|
||||
import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
|
||||
import qualified Database.SQLite.Simple as DB
|
||||
@@ -250,7 +251,6 @@ import Simplex.FileTransfer.Types
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.RetryInterval (RI2State (..))
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Common
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations (DownMigration (..), MTRError, Migration (..), MigrationsToRun (..), mtrErrorDescription)
|
||||
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
@@ -263,18 +263,25 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
|
||||
import Simplex.Messaging.Protocol
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ($>>=), (<$$>))
|
||||
import Simplex.Messaging.Util (bshow, diffToMilliseconds, eitherToMaybe, groupOn, ($>>=), (<$$>))
|
||||
import Simplex.Messaging.Version
|
||||
import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
|
||||
import System.Exit (exitFailure)
|
||||
import System.FilePath (takeDirectory)
|
||||
import System.IO (hFlush, stdout)
|
||||
import UnliftIO.Exception (onException)
|
||||
import UnliftIO.Exception (bracket, onException)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
|
||||
-- * SQLite Store implementation
|
||||
|
||||
data SQLiteStore = SQLiteStore
|
||||
{ dbFilePath :: FilePath,
|
||||
dbEncrypted :: TVar Bool,
|
||||
dbConnection :: TMVar DB.Connection,
|
||||
dbNew :: Bool
|
||||
}
|
||||
|
||||
data MigrationError
|
||||
= MEUpgrade {upMigrations :: [UpMigration]}
|
||||
| MEDowngrade {downMigrations :: [String]}
|
||||
@@ -331,35 +338,35 @@ createSQLiteStore dbFilePath dbKey migrations confirmMigrations = do
|
||||
Left e -> closeSQLiteStore st $> Left e
|
||||
|
||||
migrateSchema :: SQLiteStore -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ())
|
||||
migrateSchema st migrations confirmMigrations = do
|
||||
Migrations.initialize st
|
||||
Migrations.get st migrations >>= \case
|
||||
migrateSchema st migrations confirmMigrations = withConnection st $ \db -> do
|
||||
Migrations.initialize db
|
||||
Migrations.get db migrations >>= \case
|
||||
Left e -> do
|
||||
when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e)
|
||||
pure . Left $ MigrationError e
|
||||
Right MTRNone -> pure $ Right ()
|
||||
Right ms@(MTRUp ums)
|
||||
| dbNew st -> Migrations.run st ms $> Right ()
|
||||
| dbNew st -> Migrations.run db ms $> Right ()
|
||||
| otherwise -> case confirmMigrations of
|
||||
MCYesUp -> run ms
|
||||
MCYesUpDown -> run ms
|
||||
MCConsole -> confirm err >> run ms
|
||||
MCYesUp -> run db ms
|
||||
MCYesUpDown -> run db ms
|
||||
MCConsole -> confirm err >> run db ms
|
||||
MCError -> pure $ Left err
|
||||
where
|
||||
err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums)
|
||||
Right ms@(MTRDown dms) -> case confirmMigrations of
|
||||
MCYesUpDown -> run ms
|
||||
MCConsole -> confirm err >> run ms
|
||||
MCYesUpDown -> run db ms
|
||||
MCConsole -> confirm err >> run db ms
|
||||
MCYesUp -> pure $ Left err
|
||||
MCError -> pure $ Left err
|
||||
where
|
||||
err = MEDowngrade $ map downName dms
|
||||
where
|
||||
confirm err = confirmOrExit $ migrationErrorDescription err
|
||||
run ms = do
|
||||
run db ms = do
|
||||
let f = dbFilePath st
|
||||
copyFile f (f <> ".bak")
|
||||
Migrations.run st ms
|
||||
Migrations.run db ms
|
||||
pure $ Right ()
|
||||
|
||||
confirmOrExit :: String -> IO ()
|
||||
@@ -373,10 +380,9 @@ confirmOrExit s = do
|
||||
connectSQLiteStore :: FilePath -> String -> IO SQLiteStore
|
||||
connectSQLiteStore dbFilePath dbKey = do
|
||||
dbNew <- not <$> doesFileExist dbFilePath
|
||||
dbConn <- dbBusyLoop $ connectDB dbFilePath dbKey
|
||||
dbConnVar <- newTMVarIO dbConn
|
||||
dbConnection <- newTMVarIO =<< connectDB dbFilePath dbKey
|
||||
dbEncrypted <- newTVarIO . not $ null dbKey
|
||||
pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection = dbConnVar, dbNew}
|
||||
pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew}
|
||||
|
||||
connectDB :: FilePath -> String -> IO DB.Connection
|
||||
connectDB path key = do
|
||||
@@ -388,11 +394,13 @@ connectDB path key = do
|
||||
prepare db = do
|
||||
let exec = SQLite3.exec $ DB.connectionHandle db
|
||||
unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";"
|
||||
exec "PRAGMA busy_timeout = 100;"
|
||||
exec "PRAGMA foreign_keys = ON;"
|
||||
-- exec "PRAGMA trusted_schema = OFF;"
|
||||
exec "PRAGMA secure_delete = ON;"
|
||||
exec "PRAGMA auto_vacuum = FULL;"
|
||||
exec . fromQuery $
|
||||
[sql|
|
||||
PRAGMA foreign_keys = ON;
|
||||
-- PRAGMA trusted_schema = OFF;
|
||||
PRAGMA secure_delete = ON;
|
||||
PRAGMA auto_vacuum = FULL;
|
||||
|]
|
||||
|
||||
closeSQLiteStore :: SQLiteStore -> IO ()
|
||||
closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
|
||||
@@ -435,6 +443,37 @@ handleSQLError err e
|
||||
| DB.sqlError e == DB.ErrorConstraint = err
|
||||
| otherwise = SEInternal $ bshow e
|
||||
|
||||
withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withConnection SQLiteStore {dbConnection} =
|
||||
bracket
|
||||
(atomically $ takeTMVar dbConnection)
|
||||
(atomically . putTMVar dbConnection)
|
||||
|
||||
withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withTransaction = withTransactionCtx Nothing
|
||||
|
||||
withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000
|
||||
where
|
||||
loop :: Int -> Int -> DB.Connection -> IO a
|
||||
loop t tLim db =
|
||||
transactionWithCtx `E.catch` \(e :: SQLError) ->
|
||||
if tLim > t && DB.sqlError e == DB.ErrorBusy
|
||||
then do
|
||||
threadDelay t
|
||||
loop (t * 9 `div` 8) (tLim - t) db
|
||||
else E.throwIO e
|
||||
where
|
||||
transactionWithCtx = case ctx_ of
|
||||
Nothing -> DB.withImmediateTransaction db (action db)
|
||||
Just ctx -> do
|
||||
t1 <- getCurrentTime
|
||||
r <- DB.withImmediateTransaction db (action db)
|
||||
t2 <- getCurrentTime
|
||||
putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
|
||||
putStrLn $ "withTransactionCtx end :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
|
||||
pure r
|
||||
|
||||
createUserRecord :: DB.Connection -> IO UserId
|
||||
createUserRecord db = do
|
||||
DB.execute_ db "INSERT INTO users DEFAULT VALUES"
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
|
||||
module Simplex.Messaging.Agent.Store.SQLite.Common
|
||||
( SQLiteStore (..),
|
||||
withConnection,
|
||||
withTransaction,
|
||||
withTransactionCtx,
|
||||
dbBusyLoop,
|
||||
)
|
||||
where
|
||||
|
||||
import Control.Concurrent (threadDelay)
|
||||
import Data.Time.Clock (diffUTCTime, getCurrentTime)
|
||||
import Database.SQLite.Simple (SQLError)
|
||||
import qualified Database.SQLite.Simple as DB
|
||||
import Simplex.Messaging.Util (diffToMilliseconds)
|
||||
import UnliftIO.Exception (bracket)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
|
||||
data SQLiteStore = SQLiteStore
|
||||
{ dbFilePath :: FilePath,
|
||||
dbEncrypted :: TVar Bool,
|
||||
dbConnection :: TMVar DB.Connection,
|
||||
dbNew :: Bool
|
||||
}
|
||||
|
||||
withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withConnection SQLiteStore {dbConnection} =
|
||||
bracket
|
||||
(atomically $ takeTMVar dbConnection)
|
||||
(atomically . putTMVar dbConnection)
|
||||
|
||||
withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withTransaction = withTransactionCtx Nothing
|
||||
|
||||
withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
|
||||
withTransactionCtx ctx_ st action = withConnection st $ \db -> dbBusyLoop (transactionWithCtx db)
|
||||
where
|
||||
transactionWithCtx db = case ctx_ of
|
||||
Nothing -> DB.withImmediateTransaction db (action db)
|
||||
Just ctx -> do
|
||||
t1 <- getCurrentTime
|
||||
r <- DB.withImmediateTransaction db (action db)
|
||||
t2 <- getCurrentTime
|
||||
putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
|
||||
putStrLn $ "withTransactionCtx end :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
|
||||
pure r
|
||||
|
||||
dbBusyLoop :: forall a. IO a -> IO a
|
||||
dbBusyLoop action = loop 500 3000000
|
||||
where
|
||||
loop :: Int -> Int -> IO a
|
||||
loop t tLim =
|
||||
action `E.catch` \(e :: SQLError) ->
|
||||
if tLim > t && DB.sqlError e == DB.ErrorBusy
|
||||
then do
|
||||
threadDelay t
|
||||
loop (t * 9 `div` 8) (tLim - t)
|
||||
else E.throwIO e
|
||||
@@ -42,7 +42,6 @@ import Database.SQLite.Simple.QQ (sql)
|
||||
import qualified Database.SQLite3 as SQLite3
|
||||
import GHC.Generics (Generic)
|
||||
import Simplex.Messaging.Agent.Protocol (extraSMPServerHosts)
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Common
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
|
||||
@@ -106,42 +105,44 @@ app = sortOn name $ map migration schemaMigrations
|
||||
where
|
||||
migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
|
||||
|
||||
get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
|
||||
get st migrations = migrationsToRun migrations <$> withTransaction st getCurrent
|
||||
get :: Connection -> [Migration] -> IO (Either MTRError MigrationsToRun)
|
||||
get db migrations = migrationsToRun migrations <$> getCurrent db
|
||||
|
||||
getCurrent :: Connection -> IO [Migration]
|
||||
getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
|
||||
where
|
||||
toMigration (name, down) = Migration {name, up = "", down}
|
||||
|
||||
run :: SQLiteStore -> MigrationsToRun -> IO ()
|
||||
run st = \case
|
||||
run :: Connection -> MigrationsToRun -> IO ()
|
||||
run db = \case
|
||||
MTRUp [] -> pure ()
|
||||
MTRUp ms -> mapM_ runUp ms >> withConnection st (`execSQL` "VACUUM;")
|
||||
MTRUp ms -> mapM_ runUp ms >> execSQL "VACUUM;"
|
||||
MTRDown ms -> mapM_ runDown $ reverse ms
|
||||
MTRNone -> pure ()
|
||||
where
|
||||
runUp Migration {name, up, down} = withTransaction st $ \db -> do
|
||||
when (name == "m20220811_onion_hosts") $ updateServers db
|
||||
insert db >> execSQL db up
|
||||
runUp Migration {name, up, down} = do
|
||||
when (name == "m20220811_onion_hosts") updateServers
|
||||
DB.withImmediateTransaction db $ insert >> execSQL up
|
||||
where
|
||||
insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
|
||||
updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
|
||||
let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
|
||||
in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
|
||||
runDown DownMigration {downName, downQuery} = withTransaction st $ \db -> do
|
||||
execSQL db downQuery
|
||||
DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
|
||||
execSQL db = SQLite3.exec $ DB.connectionHandle db
|
||||
insert = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
|
||||
updateServers = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
|
||||
DB.withImmediateTransaction db $
|
||||
let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
|
||||
in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
|
||||
runDown DownMigration {downName, downQuery} = do
|
||||
DB.withImmediateTransaction db $ do
|
||||
execSQL downQuery
|
||||
DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
|
||||
execSQL = SQLite3.exec $ DB.connectionHandle db
|
||||
|
||||
initialize :: SQLiteStore -> IO ()
|
||||
initialize st = withTransaction st $ \db -> do
|
||||
initialize :: Connection -> IO ()
|
||||
initialize db = do
|
||||
cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')"
|
||||
case cs of
|
||||
[] -> createMigrations db
|
||||
[] -> createMigrations
|
||||
_ -> when ("down" `notElem` cs) $ DB.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT"
|
||||
where
|
||||
createMigrations db =
|
||||
createMigrations =
|
||||
DB.execute_
|
||||
db
|
||||
[sql|
|
||||
|
||||
Reference in New Issue
Block a user