mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-29 05:38:32 +00:00
update
This commit is contained in:
Evgeny @ SimpleX Chat
@@ -71,26 +71,29 @@ In `Simplex.FileTransfer.Server.Env` and `Simplex.FileTransfer.Server.Main`:
|
||||
|
||||
## simplexmq: server store and expiration
|
||||
|
||||
The `files` table gets a nullable `expires_at` and a `permanent BOOLEAN NOT NULL DEFAULT false`. `expires_at IS NULL` means "no explicit expiry — apply the configured default" (`created_at + ttl`); this covers legacy rows, which the migration must not re-date, since it has no access to the operator's configured TTL. `permanent = true` means the file never expires and keeps `expires_at` NULL, so a legacy row and a permanent row are distinguished by the flag, not by overloading NULL. The flag is also directly queryable for analytics.
|
||||
|
||||
Common to both stores, in `Simplex.FileTransfer.Server.Store`:
|
||||
|
||||
- add `expiresAt :: Maybe RoundedFileTime` to `FileRec`, where `Nothing` is permanent storage
|
||||
- in `createFile`, verify the proof against `sessionId <> sndKey <> digest`, resolve the requested time against the entitlement's maximum (a permanent maximum is unbounded), set `expiresAt` to the resolved expiration or `Nothing` when the result is permanent, and return it
|
||||
- add the FTTL handler, which verifies the proof against `sessionId <> sndKey <> digest`, sets `expiresAt` by the same resolution, and returns it
|
||||
- retain `created_at` for statistics and export
|
||||
- add `expiresAt :: Maybe RoundedFileTime` and `permanent :: Bool` to `FileRec`
|
||||
- in `createFile`, verify the proof against `sessionId <> sndKey <> digest`, resolve the requested time against the entitlement's maximum, and store `expiresAt`/`permanent` from the resolution (permanent when the ceiling is unbounded); return the granted storage
|
||||
- add the FTTL handler, which verifies the proof against `sessionId <> sndKey <> digest`, sets `expiresAt`/`permanent` by the same resolution, and returns it
|
||||
- `expiredFiles` takes the configured default TTL and expires a non-permanent file when `COALESCE(expiresAt, created_at + ttl) < now`
|
||||
- retain `created_at` for statistics, export, and the default-expiry fallback
|
||||
|
||||
STM store:
|
||||
|
||||
- in `expiredFiles`, select files where `expiresAt` is `Just t` and `t < now`
|
||||
- in `expiredFiles`, expire a file when `not permanent && maybe (created_at + ttl) roundedSeconds expiresAt < now`
|
||||
|
||||
PostgreSQL store, in `Simplex.FileTransfer.Server.Store.Postgres` and its migrations:
|
||||
|
||||
- add the nullable column `expires_at BIGINT`, where `NULL` is permanent storage
|
||||
- add a migration for the column and the index `idx_files_expires_at`
|
||||
- change the `expiredFiles` query to `WHERE expires_at < ? ORDER BY expires_at LIMIT ?` (a `NULL` expiration is excluded by the comparison)
|
||||
- add the nullable column `expires_at BIGINT` and `permanent BOOLEAN NOT NULL DEFAULT FALSE` (no backfill)
|
||||
- add one composite index `idx_files_expiry ON files (permanent, expires_at, created_at)`
|
||||
- `expiredFiles` query: `WHERE (NOT permanent AND expires_at < ?) OR (NOT permanent AND expires_at IS NULL AND created_at < ?) LIMIT ?` with `(now, now - ttl)`. Keep the `OR` at the top level so each disjunct is independently indexable (BitmapOr on the one composite index): `permanent` leads (equality seek skips permanent rows), `expires_at` covers arm 1's range and arm 2's `IS NULL` group, and `created_at` orders arm 2 within that group. A `COALESCE(expires_at, created_at + ttl)` predicate is avoided (not sargable, would force a sequential scan). No `ORDER BY` — the batch loop deletes all expired rows regardless of order.
|
||||
|
||||
Store log, in `Simplex.FileTransfer.Server.StoreLog`:
|
||||
|
||||
- add the optional expiration to the `AddFile` record, encoding permanent storage
|
||||
- add the `permanent` flag and the optional expiration to the `AddFile` record; a record with neither parses to `False`/`Nothing` (the configured default), never a hardcoded value
|
||||
- for older records without an expiration, default `expiresAt` to `createdAt + default storage time`
|
||||
|
||||
## simplexmq: agent
|
||||
|
||||
@@ -32,7 +32,7 @@ import Data.Int (Int64)
|
||||
import Data.List.NonEmpty (NonEmpty)
|
||||
import qualified Data.Map.Strict as M
|
||||
import qualified Data.List.NonEmpty as L
|
||||
import Data.Maybe (fromMaybe, isJust)
|
||||
import Data.Maybe (fromMaybe, isJust, isNothing)
|
||||
import qualified Data.Text as T
|
||||
import qualified Data.Text.IO as T
|
||||
import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
|
||||
@@ -495,12 +495,12 @@ processXFTPRequest sessionId HTTP2Body {bodyPart} = \case
|
||||
unless (size file `elem` sizes) $ throwE SIZE
|
||||
ts <- liftIO getFileTime
|
||||
maxSeconds <- lift $ storageMaxSeconds (xftpNewProofHeader sessionId sndKey digest) ep
|
||||
let (expiresAt, granted) = resolveStorage (roundedSeconds ts) maxSeconds storageTime
|
||||
let (expiresAt, permanent, granted) = resolveStorage (roundedSeconds ts) maxSeconds storageTime
|
||||
-- TODO validate body empty
|
||||
sId <- ExceptT $ addFileRetry st file 3 ts expiresAt
|
||||
sId <- ExceptT $ addFileRetry st file 3 ts expiresAt permanent
|
||||
rcps <- mapM (ExceptT . addRecipientRetry st 3 sId) rks
|
||||
lift $ withFileLog $ \sl -> do
|
||||
logAddFile sl sId file ts expiresAt EntityActive
|
||||
logAddFile sl sId file ts expiresAt permanent EntityActive
|
||||
logAddRecipients sl sId rcps
|
||||
stats <- asks serverStats
|
||||
lift $ incFileStat filesCreated
|
||||
@@ -513,10 +513,10 @@ processXFTPRequest sessionId HTTP2Body {bodyPart} = \case
|
||||
st <- asks fileStore
|
||||
maxSeconds <- storageMaxSeconds (xftpTimeProofHeader sessionId sId) ep
|
||||
now <- liftIO $ roundedSeconds <$> getSystemSeconds
|
||||
let (expiresAt, granted) = resolveStorage now maxSeconds storageTime
|
||||
liftIO (setFileExpiration st sId expiresAt) >>= \case
|
||||
let (expiresAt, permanent, granted) = resolveStorage now maxSeconds storageTime
|
||||
liftIO (setFileExpiration st sId expiresAt permanent) >>= \case
|
||||
Right () -> do
|
||||
withFileLog $ \sl -> logSetFileExpiration sl sId expiresAt
|
||||
withFileLog $ \sl -> logSetFileExpiration sl sId expiresAt permanent
|
||||
pure $ FRFileTime granted
|
||||
Left e -> pure $ FRErr e
|
||||
storageMaxSeconds :: BBSPresHeader -> Maybe EntitlementProof -> M s (Maybe Int64)
|
||||
@@ -529,10 +529,10 @@ processXFTPRequest sessionId HTTP2Body {bodyPart} = \case
|
||||
liftIO (verifyEntitlement entitlementIssuerKeys ph proof) >>= \case
|
||||
Just True | expiresAt > now -> pure $ fromMaybe defaultMax $ M.lookup entitlementName entCfg
|
||||
_ -> pure defaultMax
|
||||
addFileRetry :: s -> FileInfo -> Int -> RoundedFileTime -> Maybe RoundedFileTime -> M s (Either XFTPErrorType XFTPFileId)
|
||||
addFileRetry st file n ts expiresAt =
|
||||
addFileRetry :: s -> FileInfo -> Int -> RoundedFileTime -> Maybe RoundedFileTime -> Bool -> M s (Either XFTPErrorType XFTPFileId)
|
||||
addFileRetry st file n ts expiresAt permanent =
|
||||
retryAdd n $ \sId -> runExceptT $ do
|
||||
ExceptT $ addFile st sId file ts expiresAt EntityActive
|
||||
ExceptT $ addFile st sId file ts expiresAt permanent EntityActive
|
||||
pure sId
|
||||
addRecipientRetry :: s -> Int -> XFTPFileId -> RcvPublicAuthKey -> M s (Either XFTPErrorType FileRecipient)
|
||||
addRecipientRetry st n sId rpk =
|
||||
@@ -670,30 +670,34 @@ deleteOrBlockServerFile_ FileRec {filePath, fileInfo} stat storeAction = runExce
|
||||
getFileTime :: IO RoundedFileTime
|
||||
getFileTime = getRoundedSystemTime
|
||||
|
||||
resolveStorage :: Int64 -> Maybe Int64 -> FileStorageTime -> (Maybe RoundedFileTime, GrantedStorage)
|
||||
resolveStorage base maxSeconds storageTime = (expiresAt, granted)
|
||||
resolveStorage :: Int64 -> Maybe Int64 -> FileStorageTime -> (Maybe RoundedFileTime, Bool, GrantedStorage)
|
||||
resolveStorage base maxSeconds storageTime = (expiresAt, permanent, granted)
|
||||
where
|
||||
reqSeconds = case storageTime of
|
||||
FSTMax -> maxSeconds
|
||||
FSTFor hours -> Just $ let hSec = fromIntegral hours * 3600 in maybe hSec (min hSec) maxSeconds
|
||||
permanent = isNothing reqSeconds
|
||||
expiresAt = (\s -> RoundedSystemTime (base + s)) <$> reqSeconds
|
||||
granted = maybe GrantedPermanent (\(RoundedSystemTime t) -> GrantedExpires t) expiresAt
|
||||
granted = maybe GrantedPermanent (\s -> GrantedExpires (base + s)) reqSeconds
|
||||
|
||||
expireServerFiles :: FileStoreClass s => Maybe Int -> M s ()
|
||||
expireServerFiles itemDelay = do
|
||||
st <- asks fileStore
|
||||
us <- asks usedStorage
|
||||
usedStart <- readTVarIO us
|
||||
now <- liftIO $ roundedSeconds <$> getSystemSeconds
|
||||
filesCount <- liftIO $ getFileCount st
|
||||
logNote $ "Expiration check: " <> tshow filesCount <> " files"
|
||||
expireLoop st us now
|
||||
usedEnd <- readTVarIO us
|
||||
logNote $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
|
||||
expireServerFiles itemDelay =
|
||||
asks (fileExpiration . config) >>= \case
|
||||
Nothing -> pure ()
|
||||
Just ExpirationConfig {ttl = defaultTtl} -> do
|
||||
st <- asks fileStore
|
||||
us <- asks usedStorage
|
||||
usedStart <- readTVarIO us
|
||||
now <- liftIO $ roundedSeconds <$> getSystemSeconds
|
||||
filesCount <- liftIO $ getFileCount st
|
||||
logNote $ "Expiration check: " <> tshow filesCount <> " files"
|
||||
expireLoop st us now defaultTtl
|
||||
usedEnd <- readTVarIO us
|
||||
logNote $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
|
||||
where
|
||||
mbs bs = tshow (bs `div` 1048576) <> "mb"
|
||||
expireLoop st us now = do
|
||||
expired <- liftIO $ expiredFiles st now 10000
|
||||
expireLoop st us now defaultTtl = do
|
||||
expired <- liftIO $ expiredFiles st now defaultTtl 10000
|
||||
forM_ expired $ \(sId, filePath_, fileSize) -> do
|
||||
mapM_ threadDelay itemDelay
|
||||
forM_ filePath_ $ \fp ->
|
||||
@@ -706,7 +710,7 @@ expireServerFiles itemDelay = do
|
||||
unless (null sIds) $ do
|
||||
withFileLog $ \sl -> mapM_ (logDeleteFile sl) sIds
|
||||
liftIO $ deleteFiles st sIds
|
||||
expireLoop st us now
|
||||
expireLoop st us now defaultTtl
|
||||
|
||||
randomId :: Int -> M s ByteString
|
||||
randomId n = atomically . C.randomBytes n =<< asks random
|
||||
|
||||
@@ -57,6 +57,7 @@ data FileRec = FileRec
|
||||
recipientIds :: TVar (Set RecipientId),
|
||||
createdAt :: RoundedFileTime,
|
||||
expiresAt :: Maybe RoundedFileTime,
|
||||
permanent :: Bool,
|
||||
fileStatus :: TVar ServerEntityStatus
|
||||
}
|
||||
|
||||
@@ -79,9 +80,9 @@ class FileStoreClass s where
|
||||
type FileStoreConfig s
|
||||
newFileStore :: FileStoreConfig s -> IO s
|
||||
closeFileStore :: s -> IO ()
|
||||
addFile :: s -> SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> ServerEntityStatus -> IO (Either XFTPErrorType ())
|
||||
addFile :: s -> SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> Bool -> ServerEntityStatus -> IO (Either XFTPErrorType ())
|
||||
setFilePath :: s -> SenderId -> FilePath -> IO (Either XFTPErrorType ())
|
||||
setFileExpiration :: s -> SenderId -> Maybe RoundedFileTime -> IO (Either XFTPErrorType ())
|
||||
setFileExpiration :: s -> SenderId -> Maybe RoundedFileTime -> Bool -> IO (Either XFTPErrorType ())
|
||||
addRecipient :: s -> SenderId -> FileRecipient -> IO (Either XFTPErrorType ())
|
||||
deleteFile :: s -> SenderId -> IO (Either XFTPErrorType ())
|
||||
deleteFiles :: s -> [SenderId] -> IO ()
|
||||
@@ -90,7 +91,7 @@ class FileStoreClass s where
|
||||
deleteRecipient :: s -> RecipientId -> FileRec -> IO ()
|
||||
getFile :: s -> SFileParty p -> XFTPFileId -> IO (Either XFTPErrorType (FileRec, C.APublicAuthKey))
|
||||
ackFile :: s -> RecipientId -> IO (Either XFTPErrorType ())
|
||||
expiredFiles :: s -> Int64 -> Int -> IO [(SenderId, Maybe FilePath, Word32)]
|
||||
expiredFiles :: s -> Int64 -> Int64 -> Int -> IO [(SenderId, Maybe FilePath, Word32)]
|
||||
getUsedStorage :: s -> IO Int64
|
||||
getFileCount :: s -> IO Int
|
||||
|
||||
@@ -113,9 +114,9 @@ instance FileStoreClass STMFileStore where
|
||||
|
||||
closeFileStore STMFileStore {stmStoreLog} = readTVarIO stmStoreLog >>= mapM_ closeStoreLog
|
||||
|
||||
addFile STMFileStore {files} sId fileInfo createdAt expiresAt status = atomically $
|
||||
addFile STMFileStore {files} sId fileInfo createdAt expiresAt permanent status = atomically $
|
||||
ifM (TM.member sId files) (pure $ Left DUPLICATE_) $ do
|
||||
f <- newFileRec sId fileInfo createdAt expiresAt status
|
||||
f <- newFileRec sId fileInfo createdAt expiresAt permanent status
|
||||
TM.insert sId f files
|
||||
pure $ Right ()
|
||||
|
||||
@@ -130,9 +131,9 @@ instance FileStoreClass STMFileStore where
|
||||
pure $ Right ()
|
||||
_ -> pure $ Left AUTH
|
||||
|
||||
setFileExpiration STMFileStore {files} sId expiresAt = atomically $
|
||||
setFileExpiration STMFileStore {files} sId expiresAt permanent = atomically $
|
||||
TM.lookup sId files >>= \case
|
||||
Just fr -> Right () <$ TM.insert sId fr {expiresAt = expiresAt} files
|
||||
Just fr -> Right () <$ TM.insert sId fr {expiresAt = expiresAt, permanent = permanent} files
|
||||
_ -> pure $ Left AUTH
|
||||
|
||||
addRecipient st@STMFileStore {recipients} senderId (FileRecipient rId rKey) = atomically $
|
||||
@@ -177,14 +178,15 @@ instance FileStoreClass STMFileStore where
|
||||
pure $ Right ()
|
||||
_ -> pure $ Left AUTH
|
||||
|
||||
expiredFiles STMFileStore {files} now _limit = do
|
||||
expiredFiles STMFileStore {files} now defaultTtl _limit = do
|
||||
fs <- readTVarIO files
|
||||
fmap catMaybes . forM (M.toList fs) $ \(sId, FileRec {fileInfo = FileInfo {size}, filePath, expiresAt}) ->
|
||||
case expiresAt of
|
||||
Just (RoundedSystemTime t) | t < now -> do
|
||||
path <- readTVarIO filePath
|
||||
pure $ Just (sId, path, size)
|
||||
_ -> pure Nothing
|
||||
fmap catMaybes . forM (M.toList fs) $ \(sId, FileRec {fileInfo = FileInfo {size}, filePath, createdAt = RoundedSystemTime createdAt, expiresAt, permanent}) ->
|
||||
let effExpiry = maybe (createdAt + defaultTtl) roundedSeconds expiresAt
|
||||
in if not permanent && effExpiry < now
|
||||
then do
|
||||
path <- readTVarIO filePath
|
||||
pure $ Just (sId, path, size)
|
||||
else pure Nothing
|
||||
|
||||
getUsedStorage STMFileStore {files} = foldM addSize 0 =<< readTVarIO files
|
||||
where
|
||||
@@ -195,12 +197,12 @@ instance FileStoreClass STMFileStore where
|
||||
|
||||
-- Internal STM helpers
|
||||
|
||||
newFileRec :: SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> ServerEntityStatus -> STM FileRec
|
||||
newFileRec senderId fileInfo createdAt expiresAt status = do
|
||||
newFileRec :: SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> Bool -> ServerEntityStatus -> STM FileRec
|
||||
newFileRec senderId fileInfo createdAt expiresAt permanent status = do
|
||||
recipientIds <- newTVar S.empty
|
||||
filePath <- newTVar Nothing
|
||||
fileStatus <- newTVar status
|
||||
pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, fileStatus}
|
||||
pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, permanent, fileStatus}
|
||||
|
||||
withFile :: STMFileStore -> SenderId -> (FileRec -> STM (Either XFTPErrorType a)) -> STM (Either XFTPErrorType a)
|
||||
withFile STMFileStore {files} sId a =
|
||||
|
||||
@@ -82,27 +82,27 @@ instance FileStoreClass PostgresFileStore where
|
||||
closeDBStore dbStore
|
||||
mapM_ closeStoreLog dbStoreLog
|
||||
|
||||
addFile st sId fileInfo@FileInfo {sndKey, size, digest} createdAt expiresAt status =
|
||||
addFile st sId fileInfo@FileInfo {sndKey, size, digest} createdAt expiresAt permanent status =
|
||||
E.uninterruptibleMask_ $ runExceptT $ do
|
||||
void $ withDB "addFile" st $ \db ->
|
||||
E.try
|
||||
( DB.execute
|
||||
db
|
||||
"INSERT INTO files (sender_id, file_size, file_digest, sender_key, created_at, expires_at, status) VALUES (?,?,?,?,?,?,?)"
|
||||
(sId, (fromIntegral size :: Int32), Binary digest, Binary (C.encodePubKey sndKey), createdAt, expiresAt, status)
|
||||
"INSERT INTO files (sender_id, file_size, file_digest, sender_key, created_at, expires_at, permanent, status) VALUES (?,?,?,?,?,?,?,?)"
|
||||
(sId, (fromIntegral size :: Int32), Binary digest, Binary (C.encodePubKey sndKey), createdAt, expiresAt, permanent, status)
|
||||
)
|
||||
>>= either handleDuplicate (pure . Right)
|
||||
withLog "addFile" st $ \s -> logAddFile s sId fileInfo createdAt expiresAt status
|
||||
withLog "addFile" st $ \s -> logAddFile s sId fileInfo createdAt expiresAt permanent status
|
||||
|
||||
setFilePath st sId fPath = E.uninterruptibleMask_ $ runExceptT $ do
|
||||
assertUpdated $ withDB' "setFilePath" st $ \db ->
|
||||
DB.execute db "UPDATE files SET file_path = ? WHERE sender_id = ? AND file_path IS NULL AND status = 'active'" (fPath, sId)
|
||||
withLog "setFilePath" st $ \s -> logPutFile s sId fPath
|
||||
|
||||
setFileExpiration st sId expiresAt = E.uninterruptibleMask_ $ runExceptT $ do
|
||||
setFileExpiration st sId expiresAt permanent = E.uninterruptibleMask_ $ runExceptT $ do
|
||||
assertUpdated $ withDB' "setFileExpiration" st $ \db ->
|
||||
DB.execute db "UPDATE files SET expires_at = ? WHERE sender_id = ?" (expiresAt, sId)
|
||||
withLog "setFileExpiration" st $ \s -> logSetFileExpiration s sId expiresAt
|
||||
DB.execute db "UPDATE files SET expires_at = ?, permanent = ? WHERE sender_id = ?" (expiresAt, permanent, sId)
|
||||
withLog "setFileExpiration" st $ \s -> logSetFileExpiration s sId expiresAt permanent
|
||||
|
||||
addRecipient st senderId (FileRecipient rId rKey) = E.uninterruptibleMask_ $ runExceptT $ do
|
||||
void $ withDB "addRecipient" st $ \db ->
|
||||
@@ -136,13 +136,13 @@ instance FileStoreClass PostgresFileStore where
|
||||
|
||||
getFile st party fId = runExceptT $ case party of
|
||||
SFSender -> do
|
||||
row <- loadFileRow "SELECT sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, status FROM files WHERE sender_id = ?"
|
||||
row <- loadFileRow "SELECT sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, permanent, status FROM files WHERE sender_id = ?"
|
||||
fr <- ExceptT $ rowToFileRec row
|
||||
pure (fr, sndKey (fileInfo fr))
|
||||
SFRecipient -> do
|
||||
row :. Only rcpKeyBs <-
|
||||
loadFileRow
|
||||
"SELECT f.sender_id, f.file_size, f.file_digest, f.sender_key, f.file_path, f.created_at, f.expires_at, f.status, r.recipient_key FROM files f JOIN recipients r ON r.sender_id = f.sender_id WHERE r.recipient_id = ?"
|
||||
"SELECT f.sender_id, f.file_size, f.file_digest, f.sender_key, f.file_path, f.created_at, f.expires_at, f.permanent, f.status, r.recipient_key FROM files f JOIN recipients r ON r.sender_id = f.sender_id WHERE r.recipient_id = ?"
|
||||
fr <- ExceptT $ rowToFileRec row
|
||||
rcpKey <- either (const $ throwE INTERNAL) pure $ C.decodePubKey rcpKeyBs
|
||||
pure (fr, rcpKey)
|
||||
@@ -157,12 +157,12 @@ instance FileStoreClass PostgresFileStore where
|
||||
DB.execute db "DELETE FROM recipients WHERE recipient_id = ?" (Only rId)
|
||||
withLog "ackFile" st $ \s -> logAckFile s rId
|
||||
|
||||
expiredFiles st now limit =
|
||||
expiredFiles st now defaultTtl limit =
|
||||
fmap toResult $ withTransaction (dbStore st) $ \db ->
|
||||
DB.query
|
||||
db
|
||||
"SELECT sender_id, file_path, file_size FROM files WHERE expires_at < ? ORDER BY expires_at LIMIT ?"
|
||||
(now, limit)
|
||||
"SELECT sender_id, file_path, file_size FROM files WHERE (NOT permanent AND expires_at < ?) OR (NOT permanent AND expires_at IS NULL AND created_at < ?) LIMIT ?"
|
||||
(now, now - defaultTtl, limit)
|
||||
where
|
||||
toResult :: [(SenderId, Maybe FilePath, Int32)] -> [(SenderId, Maybe FilePath, Word32)]
|
||||
toResult = map (\(sId, path, size) -> (sId, path, fromIntegral size))
|
||||
@@ -179,21 +179,21 @@ instance FileStoreClass PostgresFileStore where
|
||||
|
||||
-- Internal helpers
|
||||
|
||||
mkFileRec :: SenderId -> FileInfo -> Maybe FilePath -> RoundedFileTime -> Maybe RoundedFileTime -> ServerEntityStatus -> IO FileRec
|
||||
mkFileRec senderId fileInfo path createdAt expiresAt status = do
|
||||
mkFileRec :: SenderId -> FileInfo -> Maybe FilePath -> RoundedFileTime -> Maybe RoundedFileTime -> Bool -> ServerEntityStatus -> IO FileRec
|
||||
mkFileRec senderId fileInfo path createdAt expiresAt permanent status = do
|
||||
filePath <- newTVarIO path
|
||||
recipientIds <- newTVarIO S.empty
|
||||
fileStatus <- newTVarIO status
|
||||
pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, fileStatus}
|
||||
pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, permanent, fileStatus}
|
||||
|
||||
type FileRecRow = (SenderId, Int32, ByteString, ByteString, Maybe FilePath, RoundedFileTime, Maybe RoundedFileTime, ServerEntityStatus)
|
||||
type FileRecRow = (SenderId, Int32, ByteString, ByteString, Maybe FilePath, RoundedFileTime, Maybe RoundedFileTime, Bool, ServerEntityStatus)
|
||||
|
||||
rowToFileRec :: FileRecRow -> IO (Either XFTPErrorType FileRec)
|
||||
rowToFileRec (sId, size, digest, sndKeyBs, path, createdAt, expiresAt, status) =
|
||||
rowToFileRec (sId, size, digest, sndKeyBs, path, createdAt, expiresAt, permanent, status) =
|
||||
case C.decodePubKey sndKeyBs of
|
||||
Right sndKey -> do
|
||||
let fileInfo = FileInfo {sndKey, size = fromIntegral size, digest}
|
||||
Right <$> mkFileRec sId fileInfo path createdAt expiresAt status
|
||||
Right <$> mkFileRec sId fileInfo path createdAt expiresAt permanent status
|
||||
Left _ -> pure $ Left INTERNAL
|
||||
|
||||
-- DB helpers
|
||||
@@ -248,7 +248,7 @@ importFileStore storeLogFilePath dbCfg = do
|
||||
fCnt <- withTransaction (dbStore pgStore) $ \db -> do
|
||||
DB.copy_
|
||||
db
|
||||
"COPY files (sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, status) FROM STDIN WITH (FORMAT csv)"
|
||||
"COPY files (sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, permanent, status) FROM STDIN WITH (FORMAT csv)"
|
||||
iforM_ (M.toList allFiles) $ \i (sId, fr) -> do
|
||||
DB.putCopyData db =<< fileRecToCSV sId fr
|
||||
when (i > 0 && i `mod` 10000 == 0) $ putStr (" " <> show i <> " files\r") >> hFlush stdout
|
||||
@@ -287,13 +287,13 @@ exportFileStore storeLogFilePath dbCfg = do
|
||||
!fCnt <- withTransaction (dbStore pgStore) $ \db ->
|
||||
DB.fold_
|
||||
db
|
||||
"SELECT sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, status FROM files ORDER BY created_at"
|
||||
"SELECT sender_id, file_size, file_digest, sender_key, file_path, created_at, expires_at, permanent, status FROM files ORDER BY created_at"
|
||||
(0 :: Int)
|
||||
( \(!fc) (sId, size :: Int32, digest :: ByteString, sndKeyBs :: ByteString, path :: Maybe String, createdAt, expiresAt, status) ->
|
||||
( \(!fc) (sId, size :: Int32, digest :: ByteString, sndKeyBs :: ByteString, path :: Maybe String, createdAt, expiresAt, permanent, status) ->
|
||||
case C.decodePubKey sndKeyBs of
|
||||
Right sndKey -> do
|
||||
let fileInfo = FileInfo {sndKey, size = fromIntegral size, digest}
|
||||
logAddFile sl sId fileInfo createdAt expiresAt status
|
||||
logAddFile sl sId fileInfo createdAt expiresAt permanent status
|
||||
forM_ path $ logPutFile sl sId
|
||||
pure (fc + 1)
|
||||
Left _ -> do
|
||||
@@ -331,7 +331,7 @@ iforM_ :: Monad m => [a] -> (Int -> a -> m ()) -> m ()
|
||||
iforM_ xs f = zipWithM_ f [0 ..] xs
|
||||
|
||||
fileRecToCSV :: SenderId -> FileRec -> IO ByteString
|
||||
fileRecToCSV sId FileRec {fileInfo = FileInfo {sndKey, size, digest}, filePath, createdAt, expiresAt, fileStatus} = do
|
||||
fileRecToCSV sId FileRec {fileInfo = FileInfo {sndKey, size, digest}, filePath, createdAt, expiresAt, permanent, fileStatus} = do
|
||||
path <- readTVarIO filePath
|
||||
status <- readTVarIO fileStatus
|
||||
pure $ LB.toStrict $ BB.toLazyByteString $ mconcat (BB.char7 ',' `intersperse` fields path status) <> BB.char7 '\n'
|
||||
@@ -344,6 +344,7 @@ fileRecToCSV sId FileRec {fileInfo = FileInfo {sndKey, size, digest}, filePath,
|
||||
nullable (toField <$> path),
|
||||
renderField (toField createdAt),
|
||||
nullable (toField <$> expiresAt),
|
||||
renderField (toField permanent),
|
||||
quotedField (toField status)
|
||||
]
|
||||
|
||||
|
||||
@@ -51,13 +51,14 @@ m20260823_file_expiration :: Text
|
||||
m20260823_file_expiration =
|
||||
[r|
|
||||
ALTER TABLE files ADD COLUMN expires_at BIGINT;
|
||||
UPDATE files SET expires_at = created_at + 48 * 3600;
|
||||
CREATE INDEX idx_files_expires_at ON files (expires_at);
|
||||
ALTER TABLE files ADD COLUMN permanent BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
CREATE INDEX idx_files_expiry ON files (permanent, expires_at, created_at);
|
||||
|]
|
||||
|
||||
down_m20260823_file_expiration :: Text
|
||||
down_m20260823_file_expiration =
|
||||
[r|
|
||||
DROP INDEX idx_files_expires_at;
|
||||
DROP INDEX idx_files_expiry;
|
||||
ALTER TABLE files DROP COLUMN permanent;
|
||||
ALTER TABLE files DROP COLUMN expires_at;
|
||||
|]
|
||||
|
||||
@@ -39,29 +39,30 @@ import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
|
||||
import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
|
||||
import Simplex.Messaging.Server.StoreLog
|
||||
import Simplex.Messaging.SystemTime (RoundedSystemTime (..))
|
||||
import Simplex.Messaging.Util (bshow)
|
||||
import System.IO
|
||||
|
||||
data FileStoreLogRecord
|
||||
= AddFile SenderId FileInfo RoundedFileTime (Maybe RoundedFileTime) ServerEntityStatus
|
||||
= AddFile SenderId FileInfo RoundedFileTime (Maybe RoundedFileTime) Bool ServerEntityStatus
|
||||
| PutFile SenderId FilePath
|
||||
| AddRecipients SenderId (NonEmpty FileRecipient)
|
||||
| DeleteFile SenderId
|
||||
| BlockFile SenderId BlockingInfo
|
||||
| AckFile RecipientId -- TODO add senderId as well?
|
||||
| SetFileExpiration SenderId (Maybe RoundedFileTime)
|
||||
| SetFileExpiration SenderId (Maybe RoundedFileTime) Bool
|
||||
deriving (Show)
|
||||
|
||||
instance StrEncoding FileStoreLogRecord where
|
||||
strEncode = \case
|
||||
AddFile sId file createdAt expiresAt status -> strEncode (Str "FNEW", sId, file, createdAt, status) <> " " <> maybe "P" strEncode expiresAt
|
||||
AddFile sId file createdAt expiresAt permanent status -> strEncode (Str "FNEW", sId, file, createdAt, status) <> permExpE permanent expiresAt
|
||||
PutFile sId path -> strEncode (Str "FPUT", sId, path)
|
||||
AddRecipients sId rcps -> strEncode (Str "FADD", sId, rcps)
|
||||
DeleteFile sId -> strEncode (Str "FDEL", sId)
|
||||
BlockFile sId info -> strEncode (Str "FBLK", sId, info)
|
||||
AckFile rId -> strEncode (Str "FACK", rId)
|
||||
SetFileExpiration sId expiresAt -> strEncode (Str "FTTL", sId) <> " " <> maybe "P" strEncode expiresAt
|
||||
SetFileExpiration sId expiresAt permanent -> strEncode (Str "FTTL", sId) <> permExpE permanent expiresAt
|
||||
where
|
||||
permExpE permanent expiresAt = " " <> (if permanent then "T" else "F") <> maybe "" ((" " <>) . strEncode) expiresAt
|
||||
strP =
|
||||
A.choice
|
||||
[ "FNEW " *> addFileP,
|
||||
@@ -70,7 +71,7 @@ instance StrEncoding FileStoreLogRecord where
|
||||
"FDEL " *> (DeleteFile <$> strP),
|
||||
"FBLK " *> (BlockFile <$> strP_ <*> strP),
|
||||
"FACK " *> (AckFile <$> strP),
|
||||
"FTTL " *> (SetFileExpiration <$> strP_ <*> expiryP)
|
||||
"FTTL " *> (setP <$> strP <*> permExpP)
|
||||
]
|
||||
where
|
||||
addFileP = do
|
||||
@@ -78,19 +79,23 @@ instance StrEncoding FileStoreLogRecord where
|
||||
file <- strP_
|
||||
createdAt <- strP
|
||||
status <- _strP <|> pure EntityActive
|
||||
expiresAt <- (A.space *> expiryP) <|> pure (Just $ legacyExpiry createdAt)
|
||||
pure $ AddFile sId file createdAt expiresAt status
|
||||
expiryP = (Nothing <$ A.char 'P') <|> (Just <$> strP)
|
||||
legacyExpiry (RoundedSystemTime c) = RoundedSystemTime (c + defFileExpirationHours * 3600)
|
||||
(expiresAt, permanent) <- permExpP
|
||||
pure $ AddFile sId file createdAt expiresAt permanent status
|
||||
setP sId (expiresAt, permanent) = SetFileExpiration sId expiresAt permanent
|
||||
permExpP = do
|
||||
permanent <- (A.space *> permP) <|> pure False
|
||||
expiresAt <- (A.space *> (Just <$> strP)) <|> pure Nothing
|
||||
pure (expiresAt, permanent)
|
||||
permP = (True <$ A.char 'T') <|> (False <$ A.char 'F')
|
||||
|
||||
logFileStoreRecord :: StoreLog 'WriteMode -> FileStoreLogRecord -> IO ()
|
||||
logFileStoreRecord = writeStoreLogRecord
|
||||
|
||||
logAddFile :: StoreLog 'WriteMode -> SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> ServerEntityStatus -> IO ()
|
||||
logAddFile s sId file createdAt expiresAt status = logFileStoreRecord s $ AddFile sId file createdAt expiresAt status
|
||||
logAddFile :: StoreLog 'WriteMode -> SenderId -> FileInfo -> RoundedFileTime -> Maybe RoundedFileTime -> Bool -> ServerEntityStatus -> IO ()
|
||||
logAddFile s sId file createdAt expiresAt permanent status = logFileStoreRecord s $ AddFile sId file createdAt expiresAt permanent status
|
||||
|
||||
logSetFileExpiration :: StoreLog 'WriteMode -> SenderId -> Maybe RoundedFileTime -> IO ()
|
||||
logSetFileExpiration s sId expiresAt = logFileStoreRecord s $ SetFileExpiration sId expiresAt
|
||||
logSetFileExpiration :: StoreLog 'WriteMode -> SenderId -> Maybe RoundedFileTime -> Bool -> IO ()
|
||||
logSetFileExpiration s sId expiresAt permanent = logFileStoreRecord s $ SetFileExpiration sId expiresAt permanent
|
||||
|
||||
logPutFile :: StoreLog 'WriteMode -> SenderId -> FilePath -> IO ()
|
||||
logPutFile s = logFileStoreRecord s .: PutFile
|
||||
@@ -120,15 +125,15 @@ readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.re
|
||||
Left e -> B.putStrLn $ "Log processing error (" <> bshow e <> "): " <> B.take 100 s
|
||||
_ -> pure ()
|
||||
addToStore = \case
|
||||
AddFile sId file createdAt expiresAt status
|
||||
| size file > 0 -> addFile st sId file createdAt expiresAt status
|
||||
AddFile sId file createdAt expiresAt permanent status
|
||||
| size file > 0 -> addFile st sId file createdAt expiresAt permanent status
|
||||
| otherwise -> pure $ Left SIZE
|
||||
PutFile qId path -> setFilePath st qId path
|
||||
AddRecipients sId rcps -> runExceptT $ addRecipients sId rcps
|
||||
DeleteFile sId -> deleteFile st sId
|
||||
BlockFile sId info -> blockFile st sId info True
|
||||
AckFile rId -> ackFile st rId
|
||||
SetFileExpiration sId expiresAt -> setFileExpiration st sId expiresAt
|
||||
SetFileExpiration sId expiresAt permanent -> setFileExpiration st sId expiresAt permanent
|
||||
addRecipients sId rcps = mapM_ (ExceptT . addRecipient st sId) rcps
|
||||
|
||||
writeFileStore :: StoreLog 'WriteMode -> STMFileStore -> IO ()
|
||||
@@ -137,9 +142,9 @@ writeFileStore s STMFileStore {files, recipients} = do
|
||||
readTVarIO files >>= mapM_ (logFile allRcps)
|
||||
where
|
||||
logFile :: Map RecipientId (SenderId, RcvPublicAuthKey) -> FileRec -> IO ()
|
||||
logFile allRcps FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, fileStatus} = do
|
||||
logFile allRcps FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, expiresAt, permanent, fileStatus} = do
|
||||
status <- readTVarIO fileStatus
|
||||
logAddFile s senderId fileInfo createdAt expiresAt status
|
||||
logAddFile s senderId fileInfo createdAt expiresAt permanent status
|
||||
(rcpErrs, rcps) <- M.mapEither getRcp . M.fromSet id <$> readTVarIO recipientIds
|
||||
mapM_ (logAddRecipients s senderId) $ L.nonEmpty $ M.elems rcps
|
||||
mapM_ (B.putStrLn . ("Error storing log: " <>)) rcpErrs
|
||||
|
||||
Reference in New Issue
Block a user