xftp: agent to expire rcv files older than 2 days (#699)

src/Simplex/FileTransfer/Agent.hs

@@ -70,7 +70,8 @@ startWorkers c workDir = do
   startFiles
   where
     startFiles = do
-      pendingRcvServers <- withStore' c getPendingRcvFilesServers
+      rcvFilesTTL <- asks (rcvFilesTTL . config)
+      pendingRcvServers <- withStore' c (`getPendingRcvFilesServers` rcvFilesTTL)
       forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s)
       -- start local worker for files pending decryption,
       -- no need to make an extra query for the check
@@ -140,7 +141,8 @@ runXFTPWorker c srv doWork = do
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     runXftpOperation :: m ()
     runXftpOperation = do
-      nextChunk <- withStore' c (`getNextRcvChunkToDownload` srv)
+      rcvFilesTTL <- asks (rcvFilesTTL . config)
+      nextChunk <- withStore' c $ \db -> getNextRcvChunkToDownload db srv rcvFilesTTL
       case nextChunk of
         Nothing -> noWorkToDo
         Just RcvFileChunk {rcvFileId, rcvFileEntityId, fileTmpPath, replicas = []} -> workerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) "chunk has no replicas"
@@ -207,7 +209,8 @@ runXFTPLocalWorker c doWork = do
   where
     runXftpOperation :: m ()
     runXftpOperation = do
-      nextFile <- withStore' c getNextRcvFileToDecrypt
+      rcvFilesTTL <- asks (rcvFilesTTL . config)
+      nextFile <- withStore' c (`getNextRcvFileToDecrypt` rcvFilesTTL)
       case nextFile of
         Nothing -> noWorkToDo
         Just f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath} ->

src/Simplex/Messaging/Agent.hs

@@ -1617,7 +1617,8 @@ cleanupManager c@AgentClient {subQ} = do
   forever $ do
     void . runExceptT $ do
       deleteConns `catchError` (notify "" . ERR)
-      deleteRcvFiles `catchError` (notify "" . RFERR)
+      deleteRcvFilesExpired `catchError` (notify "" . RFERR)
+      deleteRcvFilesDeleted `catchError` (notify "" . RFERR)
       deleteRcvFilesTmpPaths `catchError` (notify "" . RFERR)
     threadDelay int
   where
@@ -1625,7 +1626,13 @@ cleanupManager c@AgentClient {subQ} = do
       withLock (deleteLock c) "cleanupManager" $ do
         void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
         withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER)
-    deleteRcvFiles = do
+    deleteRcvFilesExpired = do
+      rcvFilesTTL <- asks (rcvFilesTTL . config)
+      rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
+      forM_ rcvExpired $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do
+        removePath =<< toFSFilePath p
+        withStore' c (`deleteRcvFile'` dbId)
+    deleteRcvFilesDeleted = do
       rcvDeleted <- withStore' c getCleanupRcvFilesDeleted
       forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do
         removePath =<< toFSFilePath p

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

@@ -81,6 +81,7 @@ data AgentConfig = AgentConfig
     helloTimeout :: NominalDiffTime,
     initialCleanupDelay :: Int,
     cleanupInterval :: Int,
+    rcvFilesTTL :: NominalDiffTime,
     deleteErrorCount :: Int,
     ntfCron :: Word16,
     ntfWorkerDelay :: Int,
@@ -140,6 +141,7 @@ defaultAgentConfig =
       helloTimeout = 2 * nominalDay,
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
+      rcvFilesTTL = 2 * nominalDay,
       deleteErrorCount = 10,
       ntfCron = 20, -- minutes
       ntfWorkerDelay = 100000, -- microseconds

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

@@ -146,6 +146,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     getPendingRcvFilesServers,
     getCleanupRcvFilesTmpPaths,
     getCleanupRcvFilesDeleted,
+    getRcvFilesExpired,
 
     -- * utilities
     withConnection,
@@ -180,7 +181,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 (UTCTime, getCurrentTime)
+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
 import Data.Word (Word32)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
@@ -1966,8 +1967,9 @@ deleteRcvFile' :: DB.Connection -> DBRcvFileId -> IO ()
 deleteRcvFile' db rcvFileId =
   DB.execute db "DELETE FROM rcv_files WHERE rcv_file_id = ?" (Only rcvFileId)
 
-getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> IO (Maybe RcvFileChunk)
-getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do
+getNextRcvChunkToDownload :: DB.Connection -> XFTPServer -> NominalDiffTime -> IO (Maybe RcvFileChunk)
+getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   maybeFirstRow toChunk $
     DB.query
       db
@@ -1981,11 +1983,11 @@ getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do
         JOIN rcv_files f ON f.rcv_file_id = c.rcv_file_id
         WHERE s.xftp_host = ? AND s.xftp_port = ? AND s.xftp_key_hash = ?
           AND r.received = 0 AND r.replica_number = 1
-          AND f.status = ? AND f.deleted = 0
+          AND f.status = ? AND f.deleted = 0 AND f.created_at >= ?
         ORDER BY r.created_at ASC
         LIMIT 1
       |]
-      (host, port, keyHash, RFSReceiving)
+      (host, port, keyHash, RFSReceiving, cutoffTs)
   where
     toChunk :: ((DBRcvFileId, RcvFileId, UserId, Int64, Int, FileSize Word32, FileDigest, FilePath, Maybe FilePath) :. (Int64, ChunkReplicaId, C.APrivateSignKey, Bool, Maybe Int, Int)) -> RcvFileChunk
     toChunk ((rcvFileId, rcvFileEntityId, userId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath, chunkTmpPath) :. (rcvChunkReplicaId, replicaId, replicaKey, received, delay, retries)) =
@@ -2002,8 +2004,9 @@ getNextRcvChunkToDownload db server@ProtocolServer {host, port, keyHash} = do
           replicas = [RcvFileChunkReplica {rcvChunkReplicaId, server, replicaId, replicaKey, received, delay, retries}]
         }
 
-getNextRcvFileToDecrypt :: DB.Connection -> IO (Maybe RcvFile)
-getNextRcvFileToDecrypt db = do
+getNextRcvFileToDecrypt :: DB.Connection -> NominalDiffTime -> IO (Maybe RcvFile)
+getNextRcvFileToDecrypt db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   fileId_ :: Maybe DBRcvFileId <-
     maybeFirstRow fromOnly $
       DB.query
@@ -2011,16 +2014,17 @@ getNextRcvFileToDecrypt db = do
         [sql|
           SELECT rcv_file_id
           FROM rcv_files
-          WHERE status IN (?,?) AND deleted = 0
+          WHERE status IN (?,?) AND deleted = 0 AND created_at >= ?
           ORDER BY created_at ASC LIMIT 1
         |]
-        (RFSReceived, RFSDecrypting)
+        (RFSReceived, RFSDecrypting, cutoffTs)
   case fileId_ of
     Nothing -> pure Nothing
     Just fileId -> eitherToMaybe <$> getRcvFile db fileId
 
-getPendingRcvFilesServers :: DB.Connection -> IO [XFTPServer]
-getPendingRcvFilesServers db = do
+getPendingRcvFilesServers :: DB.Connection -> NominalDiffTime -> IO [XFTPServer]
+getPendingRcvFilesServers db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   map toServer
     <$> DB.query
       db
@@ -2032,9 +2036,9 @@ getPendingRcvFilesServers db = do
         JOIN rcv_file_chunks c ON c.rcv_file_chunk_id = r.rcv_file_chunk_id
         JOIN rcv_files f ON f.rcv_file_id = c.rcv_file_id
         WHERE r.received = 0 AND r.replica_number = 1
-          AND f.status = ? AND f.deleted = 0
+          AND f.status = ? AND f.deleted = 0 AND f.created_at >= ?
       |]
-      (Only RFSReceiving)
+      (RFSReceiving, cutoffTs)
   where
     toServer :: (NonEmpty TransportHost, ServiceName, C.KeyHash) -> XFTPServer
     toServer (host, port, keyHash) = XFTPServer host port keyHash
@@ -2059,3 +2063,15 @@ getCleanupRcvFilesDeleted db =
       FROM rcv_files
       WHERE deleted = 1
     |]
+
+getRcvFilesExpired :: DB.Connection -> NominalDiffTime -> IO [(DBRcvFileId, RcvFileId, FilePath)]
+getRcvFilesExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.query
+    db
+    [sql|
+      SELECT rcv_file_id, rcv_file_entity_id, prefix_path
+      FROM rcv_files
+      WHERE created_at < ?
+    |]
+    (Only cutoffTs)