clean up rd-api

module Main (main) where

import Network.Socket.Internal (PortNumber)

import Data.String (fromString)

import System.Environment (lookupEnv)

import Data.Monoid ((<>))

import Control.Concurrent.Chan

import Control.Monad (mzero, when)

import Control.Monad.IO.Class (liftIO)

import System.Directory (setCurrentDirectory)

import qualified Data.Text as T

import Network.Wai.Handler.Warp

import Formatting

import Log

import Log.Backend.StandardOutput

-- import Network.Wai.Application.Static

import Network.Wai.Middleware.Static

import Options.Applicative

import System.Log.FastLogger

import Control.Error

import System.Exit (die)

import qualified Database.Redis as R

import qualified Text.PrettyPrint.ANSI.Leijen as D

import Config

import Utils

import Opts (argParser)

import OptsDoc (rdApiDescription)

import App (mkWaiApp)

import Worker (startWorkers)

updateRDConfigFromEnv :: RDConfig -> IO RDConfig

updateRDConfigFromEnv config = do

  webroot <- lookupEnv "WEB_ROOT"

  let newConfig =

          case webroot of

  return $ case webroot of

             Just dir -> config {webRoot=dir}

             Nothing -> config

  withSimpleStdOutLogger $ \logger -> runLogT "Main" logger $

    logInfo_ $ "webRoot is " <> T.pack (webRoot newConfig)

  return newConfig

runApiServer :: RDConfig -> IO ()

runApiServer rdConfig = withSimpleStdOutLogger $ \logger -> do

  config <- updateRDConfigFromEnv rdConfig

  conn <- R.checkedConnect $ R.defaultConnectInfo {

runApiServer :: RDConfig -> MaybeT IO ()

runApiServer rdConfig = do

  rc0 <- liftIO defaultRDRuntimeConfig

  config <- liftIO $ updateRDConfigFromEnv rdConfig

  connEi <- runExceptT $ scriptIO $ R.checkedConnect $ R.defaultConnectInfo {

            R.connectHost=redisHost config

          , R.connectPort=R.PortNumber (fromIntegral (redisPort config) :: PortNumber)

          , R.connectPort=R.PortNumber (fromIntegral (redisPort config))

          }

  fileChan <- newChan

  let runtimeConfig = RDRuntimeConfig { rcConfig=config

                                      , rcRedisConn=conn

                                      , rcFileChan=fileChan}

  startWorkers runtimeConfig

  runLogT "Main" logger $

    logInfo_ $ sformat ("will listen on " % string % ":" % int) (host config) (port config)

  conn <- case connEi of

    Left e -> do

      liftIO $ pushLogStrLn (rcLoggerSet rc0) $ toLogStr $ sformat

        ("Connect to redis at " % string % ":" % int % " failed: " % stext)

        (redisHost config) (redisPort config) e

      mzero

    Right conn ->

      return conn

  let rc = rc0 { rcConfig=config

               , rcRedisConn=conn }

  liftIO $ do

    startWorkers rc

    logl rc $ sformat ("webRoot is " % string) (webRoot config)

    logl rc $ sformat ("will listen on " % string % ":" % int) (host config) (port config)

    let warpSettings = ( setFdCacheDuration 10

                       . setFileInfoCacheDuration 10

                       . setPort (port config)

                       . setHost (fromString $ host config)) defaultSettings

  rdApi <- mkWaiApp runtimeConfig

  -- let staticApp = staticApp $ defaultFileServerSettings $ webRoot config

  -- runSettings warpSettings rdApi

  setCurrentDirectory (webRoot config)    -- static app only support serving

                                          -- from PWD

    rdApi <- mkWaiApp rc

    -- static app only support serving from PWD

    setCurrentDirectory (webRoot config)

    let app = static rdApi

    runSettings warpSettings app

rdApiDescription :: String

rdApiDescription = "rd-api is an HTTP file server that provides static file hosting and reliable\n\

\download api for rd client.\n\

\\n\

\rd-api serves files under web-root. You can use it like python3 -m http.server\n\

\\n\

\In addition, if rd command line tool is used to do the download, it will\n\

\download in a reliable way by downloading in 2MiB blocks and verify checksum\n\

\for each block.\n\

\\n\

\Usage:\n\

\    server side:\n\

\        $ ls\n\

\        bigfile1 bigfile2\n\

\        $ rd-api --host 0.0.0.0 --port 8082\n\

\\n\

\    client side:\n\

\        $ rd http://server-ip:8082/bigfile1\n\

\\n\

\Reliable download is implemented this way:\n\

\\n\

\- user uses rd client to request a resource to download.\n\

\- rd client requests resource block metadata via the /rd/ api. block metadata\n\

\  contains block count, block id, block byte offset, block content sha1sum.\n\

\- rd-api calculates and serves block metadata to rd client incrementally.\n\

\  block metadata is cached in redis after calculation.\n\

\- rd client fetches block and verifies sha1sum incrementally. When all blocks\n\

\  are downloaded and verified, combine blocks to get the final resource.\n\

\- rd client will retry on http errors and sha1sum verification failures.\n\

\- rd client supports continuing a partial download. You can press Ctrl-C to\n\

\  stop download anytime, and continue later by running the same command again."

main :: IO ()

main = runApiServer =<< execParser opts

main = do

  rdConfig <- execParser opts

  resultMaybe <- runMaybeT $ runApiServer rdConfig

  when (isNothing resultMaybe) $

      die "start rd-api failed"

  where

    opts = info (argParser <**> helper)

                (  fullDesc

module OptsDoc where

rdApiDescription :: String

rdApiDescription = "rd-api is an HTTP file server that provides static file hosting and reliable\n\

\download api for rd client.\n\

\\n\

\rd-api serves files under web-root. You can use it like python3 -m http.server\n\

\\n\

\In addition, if rd command line tool is used to do the download, it will\n\

\download in a reliable way by downloading in 2MiB blocks and verify checksum\n\

\for each block.\n\

\\n\

\Usage:\n\

\    server side:\n\

\        $ ls\n\

\        bigfile1 bigfile2\n\

\        $ rd-api --host 0.0.0.0 --port 8082\n\

\\n\

\    client side:\n\

\        $ rd http://server-ip:8082/bigfile1\n\

\\n\

\Reliable download is implemented this way:\n\

\\n\

\- user uses rd client to request a resource to download.\n\

\- rd client requests resource block metadata via the /rd/ api. block metadata\n\

\  contains block count, block id, block byte offset, block content sha1sum.\n\

\- rd-api calculates and serves block metadata to rd client incrementally.\n\

\  block metadata is cached in redis after calculation.\n\

\- rd client fetches block and verifies sha1sum incrementally. When all blocks\n\

\  are downloaded and verified, combine blocks to get the final resource.\n\

\- rd client will retry on http errors and sha1sum verification failures.\n\

\- rd client supports continuing a partial download. You can press Ctrl-C to\n\

\  stop download anytime, and continue later by running the same command again."

* current                                                             :entry:

** 

** 2018-05-08 can I make Task to support multiple getTaskResults?

- can I make Task to support multiple getTaskResults?

  when getTaskResults is called. just reset count to 0.

  remove the closed flag.

  test it.

** 2018-05-08 fix logging, clean up rd-api GET /rd/ handler.

** 2018-05-08 when start rd-api, make error obvious if connect to redis failed.

Current error is like this:

try write in MaybeT

* done                                                                :entry:

** 2018-05-08 can I make Task to support multiple getTaskResults?

- can I make Task to support multiple getTaskResults?

  when getTaskResults is called. just reset count to 0.

  remove the closed flag.

  test it.

** 2018-05-08 added some progress log. download seems sequential.

QSem based worker not effective?

name:                reliable-download

version:             0.2.0.1

version:             0.2.1.0

synopsis:            provide reliable download service via HTTP

description:         reliable-download web application and cli tool

homepage:            "https://github.com/sylecn/reliable-download#readme"

dependencies:

  - base >= 4.7 && < 5

  # project specific

  - scotty

  - wai

  - warp

  # - wai-app-static

  - wai-middleware-static

  - hedis

  - directory

  - cryptohash

  - bytestring

  - text

  - byteable

  - aeson

  - network

  - log-base

  - formatting

  - ansi-wl-pprint

  # file system

  - directory

  - filepath

  - unix

  # data types

  - bytestring

  - text

  - unordered-containers

  - aeson

  - extra

  # general utils

  - transformers

  - optparse-applicative

  - ansi-wl-pprint

  - formatting

  - fast-logger

  - errors

library:

  source-dirs: src

    - http-client

    - http-types

    - retry

    - transformers

    - socket

    - io-thread-pool

    ghc-options:

module App (mkApp, mkWaiApp) where

import Control.Monad.IO.Class (liftIO)

import Control.Monad.Trans.Class (lift)

import Data.Either (fromRight)

import Data.Either.Extra (fromRight')

import Data.Monoid ((<>))

import Data.Text.Encoding (decodeUtf8)

import Control.Concurrent.Chan

import System.IO.Error (catchIOError)

import Control.Monad (when)

import qualified Data.Text as T

import qualified Data.Text.Lazy as LT

import Data.Aeson (object, (.=))

import System.FilePath ((</>))

import System.Posix.Files (getFileStatus, fileSize)

import Control.Error

import qualified Database.Redis as R

import qualified Data.HashMap.Strict as M

import Type

import Config

import Lib (sha1sum, genBlocks)

import Utils

import qualified DB

-- | fill block sha1sum, if sha1sum is not ready yet, put "pending" there.

fillSha1sum :: RDRuntimeConfig -> FillBlockParam -> IO [BlockWithChecksum]

fillSha1sum runtimeConfig fbp = do

fillSha1sum rc fbp = do

  let hashKey = blockSha1sumHashKey fbp

  redisReply <- R.runRedis (rcRedisConn runtimeConfig) $ R.hgetall hashKey

  redisReply <- R.runRedis (rcRedisConn rc) $ R.hgetall hashKey

  case redisReply of

    Left reply -> do

      putStrLn $ "redis hgetall " <> show hashKey <> " failed: " <> show reply

      logl rc $ "redis hgetall " <> showt hashKey <> " failed: " <> showt reply

      return $ map fillBlock (fbpBlocks fbp) where

        fillBlock (blockId, start, end) = (blockId, start, end, "pending")

    Right blockIdSha1sumAlist -> do

      putStrLn $ "fillSha1sum: redis hgetall " <> show hashKey <> " ok"

      logl rc $ "fillSha1sum: redis hgetall " <> showt hashKey <> " ok"

      return $ map fillBlock (fbpBlocks fbp) where

        blockIdSha1sumMap = M.fromList blockIdSha1sumAlist

        fillBlock :: Block -> BlockWithChecksum

        fillBlock (blockId, start, end) = (blockId, start, end, decodeUtf8 $ M.lookupDefault "pending" (blockIdKey blockId) blockIdSha1sumMap)

-- | given a redis connection pool, return a Scotty app.

mkApp :: RDRuntimeConfig -> ScottyM ()

mkApp runtimeConfig = do

  get (literal "/rd/") $ json $

      object ["ok" .= True

             ,"app" .= ("reliable-download api" :: T.Text)]

  get (regex "^/rd/(.*)") $ do

    path :: T.Text <- param "1"

    let filepath = webRoot (rcConfig runtimeConfig) </> T.unpack path

    liftIO $ putStrLn $ "user request " <> filepath

    fileStatusE <- liftIO $ catchIOError

-- | given a FillBlockParam, if this file is new, send job to worker and mark

-- it as working. if there is an error, return IO Left.

processNewFileAsyncMaybe :: RDRuntimeConfig -> FillBlockParam -> ExceptT T.Text IO ()

processNewFileAsyncMaybe rc fbp = do

  let strKey = fileStatusKey fbp

  resultE <- liftIO $ DB.insertIfNotExist rc strKey fileStatusWorking

  throwOnLeft resultE

  let insertOk = fromRight' resultE

  if insertOk then liftIO $ do

      logl rc $ showt strKey <> " is a new file, sending task to worker"

      writeChan (rcFileChan rc) fbp

      return ()

  else do

    oldStatusE <- liftIO $ do

      logl rc $ showt strKey <> " is not a new file"

      -- if status is error, set it to working, then add task to fileChan

      DB.get rc strKey

    throwOnLeftMsg oldStatusE "get old file status failed"

    let oldStatus = fromRight' oldStatusE

    when (oldStatus == Just fileStatusError) $ do

        setResultE <- liftIO $ do

          logl rc $ showt strKey <> " was in " <> showt fileStatusError <> " status"

          DB.set rc strKey fileStatusWorking

        throwOnLeftMsg setResultE $ "set file status to " <> showt fileStatusWorking <> " failed"

        liftIO $ writeChan (rcFileChan rc) fbp

-- | GET /rd/.* handler

getRdHandler :: RDRuntimeConfig -> ExceptT T.Text ActionM ()

getRdHandler rc = do

  path <- lift $ param "1"

  let filepath = webRoot (rcConfig rc) </> T.unpack path

  fileStatusE <- lift $ do

    liftIO $ logl rc $ "user request " <> showt filepath

    liftIO $ catchIOError

      (fmap Right (getFileStatus filepath))

      (\e -> return $ Left $ "getFileStatus failed: " <> show e)

    case fileStatusE of

      Left errMsg -> json $ rdErrorResponse { respMsg=T.pack errMsg }  -- TODO how to do early return here?

      Right fileStatus -> do

      (\e -> do

         let msg = "getFileStatus on " <> T.pack filepath <> " failed"

         logl rc $ msg <> ":\n\t" <> T.pack (show e)

         return $ Left msg)

  throwOnLeft fileStatusE

  let fileStatus = fromRight' fileStatusE

  lift $ do

    let fileSizeInByte = toInteger $ fileSize fileStatus

        blockSizeInByte = 2097152    -- 2MiB

        blockCount = (fileSizeInByte - 1) `div` blockSizeInByte + 1

                             , fbpFileSize=fileSizeInByte

                             , fbpBlockSize=blockSizeInByte

                             , fbpBlocks=blocks }

            strKey = fileStatusKey fbp

            jsonRespFileStatusCheckFailed = json $

    resultE <- liftIO $ runExceptT $ processNewFileAsyncMaybe rc fbp

    case resultE of

      Left msg -> json $

          object ["ok" .= False

                 ,"path" .= path

                 ,"filepath" .= filepath

                     ,"msg" .= ("check file status in redis failed" :: T.Text)]

            jsonRespOk = do

              blocksWithSha1sum <- liftIO $ fillSha1sum runtimeConfig fbp

                 ,"msg" .= msg]

      Right _ -> do

          blocksWithSha1sum <- liftIO $ fillSha1sum rc fbp

          json RDResponse { respOk=True

                          , respMsg=""

                          , respPath=path

                          , respBlockCount=blockCount

                          , respBlocks=blocksWithSha1sum }

        -- if this file is new or has status "error", add task to fileChan.

        redisReply <- liftIO $ R.runRedis (rcRedisConn runtimeConfig) $ R.setnx strKey "working"

        statusCheckResult <- case redisReply of

          Left reply -> do

            liftIO $ putStrLn $ "redis setnx " <> show strKey <> " failed: " <> show reply

            return False

          Right setNxResult ->

            if setNxResult then do

                liftIO $ putStrLn $ "new file, status set to working for " <> show strKey

                liftIO $ writeChan (rcFileChan runtimeConfig) fbp

                return True

            else do

                liftIO $ putStrLn $ "not a new file, redis key " <> show strKey <> " exists"

                -- if status is error, set it to working, then add task to fileChan

                redisReply2 <- liftIO $ R.runRedis (rcRedisConn runtimeConfig) $ R.get strKey

                case redisReply2 of

                  Left reply -> do

                    liftIO $ putStrLn $ "redis file status check for " <> show strKey <> " failed: " <> show reply

                    return False

                  Right statusStr ->

                    liftIO $ if statusStr == Just "error"

                      then do

                        putStrLn $ "set file status to \"working\" for " <> show strKey

                        redisReply3 <- liftIO $ R.runRedis (rcRedisConn runtimeConfig) $ R.set strKey "working"

                        case redisReply3 of

                          Left reply -> do

                            liftIO $ putStrLn $ "set file status to \"working\" failed: " <> show reply

                            return False

                          Right _ -> do

                            liftIO $ writeChan (rcFileChan runtimeConfig) fbp

                            return True

                      else return True

        if statusCheckResult then

            jsonRespOk

        else

            jsonRespFileStatusCheckFailed

-- | given a redis connection pool, return a Scotty app.

mkApp :: RDRuntimeConfig -> ScottyM ()

mkApp rc = do

  get (literal "/rd/") $ json $

      object ["ok" .= True

             ,"app" .= ("reliable-download api" :: T.Text)]

  get (regex "^/rd/(.*)") $ do

    result <- runExceptT $ getRdHandler rc

    case result of

      Left msg -> json rdErrorResponse { respMsg=msg }

      Right resp -> return resp

  get (regex "^/test/rd/(.*)") $ do  -- for testing path capture

    path :: LT.Text <- param "1"

    let filepath = webRoot (rcConfig runtimeConfig) </> LT.unpack path

    let filepath = webRoot (rcConfig rc) </> LT.unpack path

    json $ object ["ok" .= True

                  ,"path" .= path

                  ,"filepath" .= filepath]

  get "/debug/t1" $ do

    sha1 <- liftIO $ sha1sum "/home/sylecn/persist/cache/ideaIC-2018.1.tar.gz"

    json $ object ["ok" .= True

                  ,"sha1sum" .= sha1]

  get "/debug/count" $ do

    count <- liftIO $ R.runRedis (rcRedisConn runtimeConfig) $ R.incr "count"

    count <- liftIO $ R.runRedis (rcRedisConn rc) $ R.incr "count"

    json $ object ["ok" .= True

                  ,"count" .= fromRight 0 count]