mismis-s3: Implement recursive upload

Author: erikd-ambiata

mismi-s3/mismi-s3.cabal

@@ -31,6 +31,7 @@ library
                      , conduit-extra                   == 1.1.*
                      , directory                       == 1.2.*
                      , exceptions                      >= 0.6        && < 0.9
+                     , extra                           == 1.6.*
                      , filepath                        >= 1.3        && < 1.5
                      , http-client                     >= 0.4.18     && < 0.5
                      , http-types                      == 0.8.*

mismi-s3/src/Mismi/S3/Commands.hs

@@ -24,6 +24,10 @@ module Mismi.S3.Commands (
   , uploadOrFail
   , uploadWithMode
   , uploadWithModeOrFail
+  , uploadRecursive
+  , uploadRecursiveOrFail
+  , uploadRecursiveWithMode
+  , uploadRecursiveWithModeOrFail
   , multipartUpload
   , uploadSingle
   , write
@@ -70,6 +74,7 @@ import           Control.Exception (ioError)
 import qualified Control.Exception as CE
 import           Control.Lens ((.~), (^.), to, view)
 import           Control.Monad.Catch (throwM, onException)
+import           Control.Monad.Extra (concatMapM)
 import           Control.Monad.Trans.Class (lift)
 import           Control.Monad.Trans.Resource (ResourceT, allocate, runResourceT)
 import           Control.Monad.Reader (ask)
@@ -114,10 +119,10 @@ import           P
 import           System.IO (IO, IOMode (..), SeekMode (..))
 import           System.IO (hFileSize, hSetFileSize, withFile)
 import           System.IO.Error (IOError)
-import           System.Directory (createDirectoryIfMissing, doesFileExist)
+import           System.Directory (createDirectoryIfMissing, doesFileExist, getDirectoryContents)
 import           System.FilePath (FilePath, (</>), takeDirectory)
 import           System.Posix.IO (OpenMode(..), openFd, closeFd, fdSeek, defaultFileFlags)
-import           System.Posix.Files (getFileStatus, isDirectory)
+import           System.Posix.Files (getFileStatus, isDirectory, isRegularFile)
 import qualified "unix-bytestring" System.Posix.IO.ByteString as UBS
 
 import           System.Timeout.Lifted (timeout)
@@ -365,6 +370,14 @@ upload :: FilePath -> Address -> EitherT UploadError AWS ()
 upload =
   uploadWithMode Fail
 
+uploadRecursive :: FilePath -> Address -> EitherT UploadError AWS ()
+uploadRecursive =
+  uploadRecursiveWithMode Fail
+
+uploadRecursiveOrFail :: FilePath -> Address -> AWS ()
+uploadRecursiveOrFail f a =
+  eitherT hoistUploadError pure $ uploadRecursive f a
+
 uploadOrFail :: FilePath -> Address -> AWS ()
 uploadOrFail f a =
   eitherT hoistUploadError pure $ upload f a
@@ -373,13 +386,19 @@ uploadWithModeOrFail :: WriteMode -> FilePath -> Address -> AWS ()
 uploadWithModeOrFail w f a =
   eitherT hoistUploadError pure $ uploadWithMode w f a
 
+uploadRecursiveWithModeOrFail :: WriteMode -> FilePath -> Address -> AWS ()
+uploadRecursiveWithModeOrFail w f a =
+  eitherT hoistUploadError pure $ uploadRecursiveWithMode w f a
+
 hoistUploadError :: UploadError -> AWS ()
 hoistUploadError e =
   case e of
     UploadSourceMissing f ->
       throwM $ SourceFileMissing f
     UploadDestinationExists a ->
       throwM $ DestinationAlreadyExists a
+    UploadSourceNotDirectory f ->
+      throwM $ SourceNotDirectory f
     MultipartUploadError (WorkerError a) ->
       throwM $ a
     MultipartUploadError (BlowUpError a) ->
@@ -455,6 +474,45 @@ multipartUploadWorker e mpu file a (o, c, i) =
         pure $! Right $! PartResponse i m
 
 
+uploadRecursiveWithMode :: WriteMode -> FilePath -> Address -> EitherT UploadError AWS ()
+uploadRecursiveWithMode m src (Address buck ky) = do
+  es <- tryIO $ getFileStatus src
+  case es of
+    Left _ -> left $ UploadSourceMissing src
+    Right st -> unless (isDirectory st) . left $ UploadSourceNotDirectory src
+  files <- liftIO $ listRecursivelyLocal src
+  let prefixLen = L.length (src </> "a") - 1
+      outputAddrs = fmap (\fp -> Address buck (ky // Key (T.pack $ L.drop prefixLen fp))) files
+  mapM_ (uncurry (uploadWithMode m)) $ L.zip files outputAddrs
+
+-- | Like `listRecursively` but for the local filesystem.
+listRecursivelyLocal :: MonadIO m => FilePath -> m [FilePath]
+listRecursivelyLocal topdir = do
+  entries <- liftIO $ listDirectory topdir
+  (dirs, files) <- liftIO . partitionDirsFiles $ fmap (topdir </>) entries
+  others <- concatMapM listRecursivelyLocal dirs
+  pure $ files <> others
+
+
+-- Not available with ghc 7.10 so copy it here.
+listDirectory :: FilePath -> IO [FilePath]
+listDirectory path =
+  filter f <$> getDirectoryContents path
+  where
+    f filename =
+      filename /= "." && filename /= ".."
+
+partitionDirsFiles :: MonadIO m => [FilePath] -> m ([FilePath], [FilePath])
+partitionDirsFiles =
+  pworker ([], [])
+  where
+    pworker (dirs, files) [] = pure (dirs, files)
+    pworker (dirs, files) (x:xs) = do
+      xstat <- liftIO $ getFileStatus x
+      pworker
+        (if isDirectory xstat then x : dirs else dirs, if isRegularFile xstat then x : files else files)
+        xs
+
 write :: Address -> Text -> AWS WriteResult
 write =
   writeWithMode Fail
@@ -645,9 +703,6 @@ downloadRecursiveWithMode mode src dest = do
   addrs <- lift $ listRecursively src
   mapM_ drWorker addrs
   where
-    tryIO :: MonadIO m => IO a -> m (Either IOError a)
-    tryIO = liftIO . CE.try
-
     drWorker :: Address -> EitherT DownloadError AWS ()
     drWorker addr = do
       fpdest <- hoistMaybe (DownloadInvariant addr src) $
@@ -750,3 +805,6 @@ worker input output mode env f = runEitherT . runAWST env SyncAws $ do
     (liftCopy $ copyWithMode Overwrite f out)
     (ifM (lift $ exists out) (right ()) cp)
     mode
+
+tryIO :: MonadIO m => IO a -> m (Either IOError a)
+tryIO = liftIO . CE.try

mismi-s3/src/Mismi/S3/Data.hs

@@ -76,6 +76,7 @@ data S3Error =
   | DestinationFileExists FilePath
   | DestinationNotDirectory FilePath
   | DestinationMissing FilePath
+  | SourceNotDirectory FilePath
   | AccessDenied Address
   | Invariant Text
   | Target Address Address
@@ -100,6 +101,8 @@ s3ErrorRender s3err = "[Mismi internal error] - " <> case s3err of
     "Expecting destination " <> T.pack f <> " to be a directory."
   DestinationMissing f ->
     "Download destination directory " <> T.pack f <> " does not exist."
+  SourceNotDirectory f ->
+    "Recursive upload source " <> T.pack f <> " must be a directory."
   DestinationDoesNotExist a ->
     "This address does not exist [" <> addressToText a <> "]"
   AccessDenied a ->
@@ -196,6 +199,7 @@ renderCopyError e =
 data UploadError =
     UploadSourceMissing FilePath
   | UploadDestinationExists Address
+  | UploadSourceNotDirectory FilePath
   | MultipartUploadError (RunError Error)
   deriving Show
 
@@ -206,6 +210,8 @@ renderUploadError e =
       "Can not upload when the source file does not exist [" <> T.pack f <> "]"
     UploadDestinationExists a ->
       "Can not upload when the destination object already exists [" <> addressToText a <> "]"
+    UploadSourceNotDirectory f ->
+      "Recursive upload source " <> T.pack f <> " must be a directory."
     MultipartUploadError a ->
       renderRunError a ((<>) "Multipart upload failed on a worker: " . renderError)