xftp: send progress events (#717)

src/Simplex/FileTransfer/Agent.hs

@@ -195,8 +195,8 @@ runXFTPRcvWorker c srv doWork = do
           relChunkPath = fileTmpPath </> takeFileName chunkPath
       agentXFTPDownloadChunk c userId rcvChunkId replica chunkSpec
       (complete, progress) <- withStore c $ \db -> runExceptT $ do
-        RcvFile {size = FileSize total, chunks} <-
-          ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId relChunkPath
+        liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath
+        RcvFile {size = FileSize total, chunks} <- ExceptT $ getRcvFile db rcvFileId
         let rcvd = receivedSize chunks
             complete = all chunkReceived chunks
         liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived
@@ -474,13 +474,15 @@ runXFTPSndWorker c srv doWork = do
       sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do
         updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded
         getSndFile db sndFileId
-      let complete = all chunkUploaded chunks
-      -- TODO calculate progress, notify SFPROG
+      let uploaded = uploadedSize chunks
+          total = totalSize chunks
+          complete = all chunkUploaded chunks
+      notify c sndFileEntityId $ SFPROG uploaded total
       when complete $ do
         (sndDescr, rcvDescrs) <- sndFileToDescrs sf
         notify c sndFileEntityId $ SFDONE sndDescr rcvDescrs
         forM_ prefixPath (removePath <=< toFSFilePath)
-        withStore' c $ \db -> updateSndFileStatus db sndFileId SFSComplete
+        withStore' c $ \db -> updateSndFileComplete db sndFileId
       where
         addRecipients :: SndFileChunk -> SndFileChunkReplica -> m SndFileChunkReplica
         addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {rcvIdsKeys}
@@ -547,6 +549,13 @@ runXFTPSndWorker c srv doWork = do
                   Just ch@FileChunk {replicas} -> ch {replicas = replica' : replicas}
                   _ -> FileChunk {chunkNo, digest, chunkSize, replicas = [replica']}
                 replica' = FileChunkReplica {server, replicaId, replicaKey}
+        uploadedSize :: [SndFileChunk] -> Int64
+        uploadedSize = foldl' (\sz ch -> sz + uploadedChunkSize ch) 0
+        uploadedChunkSize ch
+          | chunkUploaded ch = fromIntegral (sndChunkSize ch)
+          | otherwise = 0
+        totalSize :: [SndFileChunk] -> Int64
+        totalSize = foldl' (\sz ch -> sz + fromIntegral (sndChunkSize ch)) 0
         chunkUploaded :: SndFileChunk -> Bool
         chunkUploaded SndFileChunk {replicas} =
           any (\SndFileChunkReplica {replicaStatus} -> replicaStatus == SFRSUploaded) replicas

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

@@ -134,6 +134,7 @@ module Simplex.Messaging.Agent.Store.SQLite
 
     -- Rcv files
     createRcvFile,
+    getRcvFile,
     getRcvFileByEntityId,
     updateRcvChunkReplicaDelay,
     updateRcvFileChunkReceived,
@@ -1947,12 +1948,11 @@ updateRcvChunkReplicaDelay db replicaId delay = do
   updatedAt <- getCurrentTime
   DB.execute db "UPDATE rcv_file_chunk_replicas SET delay = ?, retries = retries + 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (delay, updatedAt, replicaId)
 
-updateRcvFileChunkReceived :: DB.Connection -> Int64 -> Int64 -> DBRcvFileId -> FilePath -> IO (Either StoreError RcvFile)
-updateRcvFileChunkReceived db rId cId fId chunkTmpPath = do
+updateRcvFileChunkReceived :: DB.Connection -> Int64 -> Int64 -> FilePath -> IO ()
+updateRcvFileChunkReceived db replicaId chunkId chunkTmpPath = do
   updatedAt <- getCurrentTime
-  DB.execute db "UPDATE rcv_file_chunk_replicas SET received = 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (updatedAt, rId)
-  DB.execute db "UPDATE rcv_file_chunks SET tmp_path = ?, updated_at = ? WHERE rcv_file_chunk_id = ?" (chunkTmpPath, updatedAt, cId)
-  getRcvFile db fId
+  DB.execute db "UPDATE rcv_file_chunk_replicas SET received = 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (updatedAt, replicaId)
+  DB.execute db "UPDATE rcv_file_chunks SET tmp_path = ?, updated_at = ? WHERE rcv_file_chunk_id = ?" (chunkTmpPath, updatedAt, chunkId)
 
 updateRcvFileStatus :: DB.Connection -> DBRcvFileId -> RcvFileStatus -> IO ()
 updateRcvFileStatus db rcvFileId status = do

tests/XFTPAgent.hs

@@ -81,7 +81,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do
   rfd <- runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     sfId <- xftpSendFile sndr 1 filePath 2
-    -- sfProgress sndr $ mb 18
+    sfProgress sndr $ mb 18
     ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr
     liftIO $ sfId' `shouldBe` sfId
     pure rfd1
@@ -121,7 +121,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
     runRight $ do
       xftpStartWorkers sndr (Just senderFiles)
       sfId <- xftpSendFile sndr 1 filePath 2
-      -- sfProgress sndr $ mb 18
+      sfProgress sndr $ mb 18
       ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr
       liftIO $ sfId' `shouldBe` sfId
       pure rfd1
@@ -166,7 +166,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
     runRight $ do
       xftpStartWorkers sndr (Just senderFiles)
       sfId <- xftpSendFile sndr 1 filePath 2
-      -- sfProgress sndr $ mb 18
+      sfProgress sndr $ mb 18
       ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr
       liftIO $ sfId' `shouldBe` sfId
       pure rfd1