nuthatch/src/Nuthatch/App.hs

module Nuthatch.App
  ( Env (..)
  , newEnv
  , application
  , utf8Boundary
  , authenticationHandler
  , loginResponse
  , dispatchAndFinalize
  , collectKeysBoundedWith
  , PutFailure (..)
  , runPut
  , runDelete
  , runRecovery
  ) where

import           Control.Concurrent (threadDelay)
import           Control.Concurrent.Async (mapConcurrently_)
import           Control.Exception
  ( SomeException
  , throwIO
  , try
  )
import           Control.Monad (forM, forever, when)
import           Control.Monad.IO.Class (liftIO)
import           Control.Retry
  ( retrying )
import qualified Data.Aeson as A
import           Data.Bifunctor (first)
import qualified Data.ByteString as BS
import qualified Data.ByteString.Lazy as BL
import           Data.Either (isRight)
import           Data.Maybe (fromMaybe)
import           Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as TE
import qualified Data.Text.IO as TIO
import           Data.Word (Word64)
import           Database.PostgreSQL.Simple (Only (..))
import           GHC.Clock (getMonotonicTimeNSec)
import           Network.HTTP.Types
  ( hContentType
  , hCookie
  , hLocation
  , methodGet
  , status200
  , status400
  , status404
  , status405
  , urlDecode
  )
import qualified Network.HTTP.Types.URI as URI
import           Network.Wai (Request (..), responseFile, responseLBS)
import           Nuthatch.Api
import           Nuthatch.Audit (audit)
import           Nuthatch.Config (Config (..))
import           Nuthatch.Db (Db, ensurePersonalPool, withTransaction)
import qualified Nuthatch.Db as Db
import           Nuthatch.Oidc
import           Nuthatch.Paths
import           Nuthatch.Protocol
import           Nuthatch.Redis
  ( Redis
  , cacheGet
  , cacheInvalidatePrefixes
  , cacheSet
  , throttleCheck
  , throttleTake
  )
import           Nuthatch.Recovery
  ( recoveryRetryPolicy
  )
import           Nuthatch.S3
  ( ListPage (..)
  , ObjectInfo (..)
  , PutObjectError (..)
  , S3
  , S3Error (..)
  , deleteObject
  , getObjectSource
  , headObject
  , listObjects
  , objectStoreAvailable
  , observeObjectFully
  , putObjectStream
  , renderS3Error
  )
import           Nuthatch.Session
import           Nuthatch.Streaming
  ( BandwidthError (..)
  , UploadStreamError (..)
  )
import           Nuthatch.Types
import           Nuthatch.Ui (browserScript, productPage)
import           Servant
import           Servant.Server.Experimental.Auth
  ( AuthHandler
  , mkAuthHandler
  )
import           Servant.Types.SourceT (SourceT)
import           System.Directory (doesFileExist)
import           System.Timeout (timeout)
import           Web.Cookie
  ( defaultSetCookie
  , parseCookies
  , renderSetCookieBS
  , sameSiteLax
  , setCookieHttpOnly
  , setCookieMaxAge
  , setCookieName
  , setCookiePath
  , setCookieSameSite
  , setCookieSecure
  , setCookieValue
  )

data Env = Env
  { envConfig :: Config
  , envDb     :: Db
  , envRedis  :: Redis
  , envS3     :: S3
  }

newEnv :: Config -> Db -> Redis -> S3 -> Env
newEnv = Env

application :: Env -> Application
application env =
  utf8Boundary $
    serveWithContext
      api
      (authenticationHandler (envConfig env) :. EmptyContext)
      (server env)

utf8Boundary :: Application -> Application
utf8Boundary downstream request sendResponse
  | requestUtf8Valid request = downstream request sendResponse
  | otherwise =
      sendResponse $
        responseLBS status400
          [(hContentType, "application/json")]
          (A.encode InvalidUtf8)

requestUtf8Valid :: Request -> Bool
requestUtf8Valid request =
  validUtf8 (urlDecode False (rawPathInfo request))
    && all validQueryItem (queryString request)
  where
    validQueryItem (name, value) =
      validUtf8 name && maybe True validUtf8 value
    validUtf8 = isRight . TE.decodeUtf8'

server :: Env -> Server Api
server env =
       pure (HealthResponse True)
  :<|> handleReady env
  :<|> pure (BrowserScript browserScript)
  :<|> handleAppRequest
  :<|> staticAssetsServer
  :<|> handleLoginRequest env
  :<|> handleOidcCallback env
  :<|> handleLogoutRequest env
  :<|> handleBrowseRequest env
  :<|> handleDownloadRequest env
  :<|> handleUploadRequest env
  :<|> handleDeleteRequest env
  :<|> handleDeletePrefixRequest env
  :<|>
    (    handleSessionRequest
    :<|> handleBrowseRequest env
    :<|> handleDownloadRequest env
    :<|> handleUploadRequest env
    :<|> handleDeleteRequest env
    :<|> handleDeletePlanRequest env
    :<|> handleDeleteBatchRequest env
    )

handleAppRequest
  :: SessionContext
  -> Handler (Headers '[Header "Content-Security-Policy" Text] HtmlPage)
handleAppRequest _ = do
  page <- liftIO (TIO.readFile "web/dist/index.html")
  pure (addHeader applicationContentSecurityPolicy (HtmlPage page))

handleSessionRequest :: SessionContext -> Handler SessionResponse
handleSessionRequest session =
  case sessionCsrf session of
    Nothing -> throwOperationError err403 AuthenticationRequired
    Just csrf ->
      let user = sessionUser session
      in pure SessionResponse
        { sessionResponseUserId = suUserId user
        , sessionResponseDisplay = suDisplay user
        , sessionResponseCsrf = csrf
        }

applicationContentSecurityPolicy :: Text
applicationContentSecurityPolicy =
  "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; \
  \base-uri 'none'; form-action 'self'; frame-ancestors 'none'; object-src 'none'"

staticAssetsServer :: SessionContext -> Server Raw
staticAssetsServer _ =
  Tagged $ \request sendResponse ->
    case (requestMethod request, pathInfo request) of
      (method, [asset])
        | method == methodGet
        , validAssetName asset -> do
            let path = "web/dist/assets/" <> T.unpack asset
            exists <- doesFileExist path
            if exists
              then sendResponse $
                responseFile
                  status200
                  [ (hContentType, assetContentType asset)
                  , ("X-Content-Type-Options", "nosniff")
                  , ("Cache-Control", "public, max-age=31536000, immutable")
                  ]
                  path
                  Nothing
              else sendResponse (responseLBS status404 [] "")
      (method, _) | method /= methodGet ->
        sendResponse (responseLBS status405 [] "")
      _ ->
        sendResponse (responseLBS status404 [] "")
  where
    validAssetName asset =
      not (T.null asset)
        && T.all
          (\character ->
            character >= 'a' && character <= 'z'
              || character >= 'A' && character <= 'Z'
              || character >= '0' && character <= '9'
              || character `elem` (".-_" :: String))
          asset
    assetContentType asset
      | ".js" `T.isSuffixOf` asset = "text/javascript; charset=utf-8"
      | ".css" `T.isSuffixOf` asset = "text/css; charset=utf-8"
      | otherwise = "application/octet-stream"

authenticationHandler :: Config -> AuthHandler Request SessionContext
authenticationHandler cfg = mkAuthHandler $ \request -> do
  let cookies = requestCookies request
      sessionName = TE.encodeUtf8 (cfgCookieName cfg)
      csrfName = TE.encodeUtf8 (cfgCsrfCookieName cfg)
  token <- maybe (throwError loginRequired) pure (lookup sessionName cookies)
  sessionToken <- parseCookieValue validateSessionToken token
  verified <- liftIO (verifySession cfg sessionToken)
  user <- either (const (throwError loginRequired)) pure verified
  csrf <- traverse
    (parseCookieValue validateCsrfToken)
    (lookup csrfName cookies)
  pure SessionContext
    { sessionUser = user
    , sessionCsrf = csrf
    }
  where
    loginRequired = err303
      { errHeaders = [(hLocation, "/login")]
      , errBody = A.encode AuthenticationRequired
      }
    parseCookieValue validate bytes =
      case TE.decodeUtf8' bytes of
        Left _ -> throwOperationError err400 MalformedAuthentication
        Right value ->
          either
            (const (throwOperationError err400 MalformedAuthentication))
            pure
            (validate value)

handleReady :: Env -> Handler ReadyResponse
handleReady env = do
  pg <- liftIO $
    try (Db.query_ (envDb env) "SELECT 1") :: Handler (Either SomeException [Only Int])
  case pg of
    Right _ -> pure (ReadyResponse True)
    Left _ -> throwOperationError err503 ReadyDependencyUnavailable

handleLoginRequest
  :: Env
  -> Handler LoginStart
handleLoginRequest env = do
  let cfg = envConfig env
  authorizationResult <- liftIO (beginAuthorization cfg)
  authorization <- either
    (const (throwOperationError err502 OidcAuthorizationFailed))
    pure
    authorizationResult
  pure $
    addHeader
      (cookieValue
        (cfgOidcTransactionCookieName cfg)
        (authorizationTransaction authorization)
        "/oidc/callback"
        True) $
      addHeader (authorizationLocation authorization) NoContent

handleOidcCallback
  :: Env
  -> Text
  -> Text
  -> Text
  -> Handler LoginSuccess
handleOidcCallback env code state cookieHeader = do
  let cfg = envConfig env
      cookies = parseCookies (TE.encodeUtf8 cookieHeader)
      transactionName = TE.encodeUtf8 (cfgOidcTransactionCookieName cfg)
  transaction <- maybe
    (throwOperationError err400 OidcCallbackRejected)
    (parseCookieValueText OidcCallbackRejected)
    (lookup transactionName cookies)
  identityResult <- liftIO (redeemAuthorizationCode cfg transaction state code)
  identity <- case identityResult of
    Left reason -> do
      liftIO $ audit "login" "rejected" Nothing [("reason", A.String reason)]
      throwOperationError err400 OidcCallbackRejected
    Right accepted -> pure accepted
  let uid = encodeUserId (identityIssuer identity) (identitySubject identity)
  liftIO (ensurePersonalPool (envDb env) cfg uid)
  tokenResult <- liftIO $
    mintSession
      cfg
      (identityIssuer identity)
      (identitySubject identity)
      (identityDisplay identity)
  token <- either
    (const (throwOperationError err500 (SessionCreationFailed "session creation failed")))
    pure
    tokenResult
  csrf <- liftIO newCsrfToken
  liftIO $ audit "login" "ok" (Just (unUserId uid)) []
  pure (loginResponse cfg token csrf)

loginResponse :: Config -> SessionToken -> CsrfToken -> LoginSuccess
loginResponse cfg token csrf =
  addHeader (cookieValue (cfgCookieName cfg) (unSessionToken token) "/" True) $
    addHeader (cookieValue (cfgCsrfCookieName cfg) (unCsrfToken csrf) "/" False) $
      addHeader
        (expiredCookieValue
          (cfgOidcTransactionCookieName cfg)
          "/oidc/callback") $
        addHeader "/app" NoContent

handleLogoutRequest
  :: Env
  -> SessionContext
  -> CsrfToken
  -> Handler LogoutSuccess
handleLogoutRequest env session providedCsrf = do
  requireCsrf LogoutCsrfRejected session providedCsrf
  let cfg = envConfig env
  pure $
    addHeader (expiredCookieValue (cfgCookieName cfg) "/") $
      addHeader (expiredCookieValue (cfgCsrfCookieName cfg) "/") $
        addHeader
          (expiredCookieValue
            (cfgOidcTransactionCookieName cfg)
            "/oidc/callback")
          NoContent

handleBrowseRequest
  :: Env
  -> SessionContext
  -> Maybe Root
  -> Maybe BrowsePrefix
  -> Maybe SearchQuery
  -> Maybe PaginationToken
  -> Handler BrowseView
handleBrowseRequest env session selectedRoot selectedPrefix search continuation = do
  let user = sessionUser session
      root = fromMaybe Personal selectedRoot
      prefix = maybe "" browsePrefixText selectedPrefix
  liftIO (ensurePersonalPool (envDb env) (envConfig env) (suUserId user))
  result <- liftIO $ case search of
    Nothing -> do
      listed <- listEntries env user root prefix
        (unPaginationToken <$> continuation)
      pure (fmap (\(entries, next) -> (entries, next, False)) listed)
    Just query -> do
      found <- searchEntries env user root (unSearchQuery query)
      pure (fmap (\(entries, partial) -> (entries, Nothing, partial)) found)
  (entries, nextContinuation, partial) <- either
    (throwOperationError status502Error . BrowseObjectStoreFailed)
    pure
    result
  let response = BrowseResponse
        { browseEntries = entries
        , browseContinuation = PaginationToken <$> nextContinuation
        , browsePartial = partial
        }
  pure BrowseView
    { browseViewHtml = HtmlPage (productPage entries)
    , browseViewJson = response
    }

handleDownloadRequest
  :: Env
  -> SessionContext
  -> Root
  -> RelativePath
  -> Maybe Text
  -> Handler DownloadSuccess
handleDownloadRequest env session root path requestedRange = do
  when (requestedRange /= Nothing) $
    throwOperationError err416 DownloadRangeUnsupported
  let user = sessionUser session
      key = buildObjectKey root (suUserId user) path
  metadata <- liftIO (headObject (envS3 env) key)
  info <- case metadata of
    Left message -> throwOperationError status502Error
      (DownloadObjectStoreFailed (renderS3Error message))
    Right Nothing -> throwOperationError err404 DownloadNotFound
    Right (Just value) -> pure value
  requireDownloadBandwidth env user (oiSize info)
  let inspect = inspectBandwidth env user
      body = getObjectSource (envS3 env) key (transferPieceSize env) inspect
  liftIO $ audit "download" "ok" (Just (unUserId (suUserId user)))
    [ ("root", A.toJSON (rootLabel root))
    , ("bytes", A.toJSON (oiSize info))
    ]
  let filename = basename path
  pure $
    addHeader (contentDisposition filename) $
      addHeader "nosniff" $
        addHeader "sandbox" $
          addHeader (oiSize info) body

handleUploadRequest
  :: Env
  -> SessionContext
  -> Root
  -> RelativePath
  -> OperationId
  -> DeclaredSize
  -> ContentSha256
  -> CsrfToken
  -> SourceIO BS.ByteString
  -> Handler UploadResponse
handleUploadRequest env session root path operation declaredSize contentIdentity providedCsrf bodySource = do
  requireCsrf UploadCsrfRejected session providedCsrf
  let user = sessionUser session
      size = unDeclaredSize declaredSize
      configuredMaximum = cfgMaxObjectBytes (envConfig env)
  when (size > configuredMaximum) $
    throwOperationError err413 UploadTooLarge
  requireUploadBandwidth env user size
  outcome <- liftIO $
    runPut env user root path operation size contentIdentity bodySource
  result <- either
    throwPutFailure
    pure
    outcome
  liftIO $ audit "put" (mutationAuditResult result) (Just (unUserId (suUserId user)))
    [ ("root", A.toJSON (rootLabel root))
    , ("bytes", A.toJSON size)
    , ("operation_id", A.toJSON (unOperationId operation))
    ]
  pure UploadResponse
    { uploadResponseOperationId = operation
    , uploadResponseResult = result
    , uploadResponseIdempotencyWindowSeconds =
        cfgIdempotencyWindowSec (envConfig env)
    }

requireDownloadBandwidth :: Env -> SessionUser -> Integer -> Handler ()
requireDownloadBandwidth env user size = do
  available <- liftIO $
    throttleCheck
      (envRedis env)
      (envConfig env)
      (suUserId user)
      (min size (toInteger (transferPieceSize env)))
  case available of
    Left _ -> throwOperationError err503 DownloadThrottleUnavailable
    Right False -> throwOperationError err429 DownloadBandwidthExceeded
    Right True -> pure ()

requireUploadBandwidth :: Env -> SessionUser -> Integer -> Handler ()
requireUploadBandwidth env user size = do
  available <- liftIO $
    throttleCheck
      (envRedis env)
      (envConfig env)
      (suUserId user)
      (min size (toInteger (transferPieceSize env)))
  case available of
    Left _ -> throwOperationError err503 UploadThrottleUnavailable
    Right False -> throwOperationError err429 UploadBandwidthExceeded
    Right True -> pure ()

inspectBandwidth :: Env -> SessionUser -> BS.ByteString -> IO ()
inspectBandwidth env user chunk = do
  result <- throttleTake
    (envRedis env)
    (envConfig env)
    (suUserId user)
    (toInteger (BS.length chunk))
  case result of
    Left _ -> throwIO BandwidthUnavailable
    Right False -> throwIO BandwidthExceeded
    Right True -> pure ()

transferPieceSize :: Env -> Int
transferPieceSize env =
  fromInteger (max 1 (min 65536 (cfgThrottleBurstBytes (envConfig env))))

handleDeleteRequest
  :: Env
  -> SessionContext
  -> DeleteRequest
  -> Handler DeleteResponse
handleDeleteRequest env session request = do
  requireCsrf DeleteCsrfRejected session (deleteCsrf request)
  let user = sessionUser session
      operation = deleteOperationId request
      root = deleteRoot request
  outcome <- liftIO $
    runDelete env user root (deletePath request) operation
  result <- either
    (throwOperationError err409 . DeleteMutationRejected)
    pure
    outcome
  liftIO $ audit "delete" (mutationAuditResult result)
    (Just (unUserId (suUserId user)))
    [ ("root", A.toJSON (rootLabel root))
    , ("operation_id", A.toJSON (unOperationId operation))
    ]
  pure DeleteResponse
    { deleteResponseOperationId = operation
    , deleteResponseResult = result
    , deleteResponseIdempotencyWindowSeconds =
        cfgIdempotencyWindowSec (envConfig env)
    }

handleDeletePrefixRequest
  :: Env
  -> SessionContext
  -> DeletePrefixRequest
  -> Handler DeletePrefixResponse
handleDeletePrefixRequest _env session request = do
  requireCsrf DeletePrefixCsrfRejected session (deletePrefixCsrf request)
  throwOperationError err409
    (DeletePrefixMutationRejected
      "use delete-plan and delete-batch with client operation ids")

handleDeletePlanRequest
  :: Env
  -> SessionContext
  -> Root
  -> RelativePath
  -> Maybe PaginationToken
  -> Handler DeletePlanResponse
handleDeletePlanRequest env session root path continuation = do
  let user = sessionUser session
      rootPrefix = case root of
        Personal -> "users/" <> unUserId (suUserId user) <> "/"
        Shared   -> "shared/"
      objectPrefix = rootPrefix <> unRelativePath path <> "/"
  listed <- liftIO $
    listObjects
      (envS3 env)
      objectPrefix
      Nothing
      (unPaginationToken <$> continuation)
      (cfgDeletePageSize (envConfig env))
  page <- either
    (throwOperationError status502Error
      . DeletePrefixMutationRejected
      . renderS3Error)
    pure
    listed
  paths <- either
    (throwOperationError err409 . DeletePrefixMutationRejected)
    pure
    (traverse
      (validateRelativePath . T.drop (T.length rootPrefix) . fst)
      (lpObjects page))
  pure DeletePlanResponse
    { deletePlanPaths = paths
    , deletePlanContinuation = PaginationToken <$> lpContinuationToken page
    }

handleDeleteBatchRequest
  :: Env
  -> SessionContext
  -> DeleteBatchRequest
  -> Handler DeleteBatchResponse
handleDeleteBatchRequest env session request = do
  requireCsrf DeletePrefixCsrfRejected session (deleteBatchCsrf request)
  let items = deleteBatchItems request
  when (null items || length items > cfgDeletePageSize (envConfig env)) $
    throwOperationError err409
      (DeletePrefixMutationRejected "delete batch size is invalid")
  results <- liftIO $ forM items $ \item -> do
    outcome <- runDelete
      env
      (sessionUser session)
      (deleteBatchRoot request)
      (deleteBatchPath item)
      (deleteBatchOperationId item)
    pure $ case outcome of
      Left message -> DeleteItemResponse
        { deleteItemPath = deleteBatchPath item
        , deleteItemOperationId = deleteBatchOperationId item
        , deleteItemResult = Nothing
        , deleteItemError = Just message
        }
      Right result -> DeleteItemResponse
        { deleteItemPath = deleteBatchPath item
        , deleteItemOperationId = deleteBatchOperationId item
        , deleteItemResult = Just result
        , deleteItemError = Nothing
        }
  pure (DeleteBatchResponse results)

requireCsrf :: A.ToJSON error => error -> SessionContext -> CsrfToken -> Handler ()
requireCsrf csrfError session provided =
  case sessionCsrf session of
    Just expected | constantTimeEq expected provided -> pure ()
    _ -> throwOperationError err403 csrfError

throwOperationError :: A.ToJSON error => ServerError -> error -> Handler a
throwOperationError serverError operationError =
  throwError serverError
    { errBody = A.encode operationError
    , errHeaders = ("Content-Type", "application/json") : errHeaders serverError
    }

status502Error :: ServerError
status502Error = ServerError
  { errHTTPCode = 502
  , errReasonPhrase = "Bad Gateway"
  , errBody = ""
  , errHeaders = []
  }

browsePrefixText :: BrowsePrefix -> Text
browsePrefixText (BrowsePrefix Nothing) = ""
browsePrefixText (BrowsePrefix (Just path)) = unRelativePath path

cookieValue :: Text -> Text -> BS.ByteString -> Bool -> Text
cookieValue name value path httpOnly =
  TE.decodeUtf8 $
    renderSetCookieBS $
      defaultSetCookie
        { setCookieName = TE.encodeUtf8 name
        , setCookieValue = TE.encodeUtf8 value
        , setCookiePath = Just path
        , setCookieHttpOnly = httpOnly
        , setCookieSecure = True
        , setCookieSameSite = Just sameSiteLax
        }

expiredCookieValue :: Text -> BS.ByteString -> Text
expiredCookieValue name path =
  TE.decodeUtf8 $
    renderSetCookieBS $
      defaultSetCookie
        { setCookieName = TE.encodeUtf8 name
        , setCookieValue = ""
        , setCookiePath = Just path
        , setCookieMaxAge = Just 0
        , setCookieHttpOnly = True
        , setCookieSecure = True
        , setCookieSameSite = Just sameSiteLax
        }

mutationAuditResult :: MutationResult -> Text
mutationAuditResult MutationSucceeded = "ok"
mutationAuditResult MutationPending   = "pending"
mutationAuditResult MutationFailed    = "failed"
mutationAuditResult MutationRejected  = "rejected"
mutationAuditResult MutationConflict  = "conflict"

contentDisposition :: Text -> Text
contentDisposition filename =
  "attachment; filename=\"" <> fallback <> "\"; filename*=UTF-8''" <> encoded
  where
    fallback = T.map safeAscii filename
    encoded = TE.decodeUtf8 (URI.urlEncode True (TE.encodeUtf8 filename))
    safeAscii character
      | character >= ' ' && character <= '~'
          && character `notElem` ("\"\\;" :: String) = character
      | otherwise = '_'

runPut
  :: Env
  -> SessionUser
  -> Root
  -> RelativePath
  -> OperationId
  -> Integer
  -> ContentSha256
  -> SourceT IO BS.ByteString
  -> IO (Either PutFailure MutationResult)
runPut env user root rel op size sha body = do
  ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
  let uid = suUserId user
      key = buildObjectKey root uid rel
      pool = case root of
        Personal -> personalPool uid
        Shared   -> sharedPool
  admitted <- withTransaction (envDb env) $ \conn -> do
    existing <- mutationEpochExists
      conn uid Put op (cfgIdempotencyWindowSec (envConfig env))
    available <-
      if existing then pure True else objectStoreAvailable (envS3 env)
    if available
      then Just <$> admitMutation conn uid Put op pool key rel size (Just sha)
        (cfgMaxObjectBytes (envConfig env))
        (cfgIdempotencyWindowSec (envConfig env))
      else pure Nothing
  case admitted of
    Nothing ->
      pure (Left (PutDispatchFailed (PutObjectStoreError S3TransportError)))
    Just outcome ->
      case outcome of
        AdmitConflict -> pure (Right MutationConflict)
        AdmitRejected _ _ -> pure (Right MutationRejected)
        AdmitReplay MutationPending _ -> pure (Right MutationPending)
        AdmitReplay result _ -> pure (Right result)
        AdmitPrepared rec ->
          dispatchAndFinalize
            env
            user
            root
            rec
            (PutDispatchFailed (PutObjectStoreError S3TransportError))
            (putObjectStream
              (envS3 env)
              key
              size
              (Just sha)
              (transferPieceSize env)
              (inspectBandwidth env user)
              body
              >>= pure . either (Left . PutDispatchFailed) Right)

newtype PutFailure
  = PutDispatchFailed PutObjectError
  deriving stock (Eq, Show)

throwPutFailure :: PutFailure -> Handler a
throwPutFailure failure =
  case failure of
    PutDispatchFailed (PutObjectStreamError streamError) ->
      case streamError of
        UploadStreamInvalid message ->
          throwOperationError err400 (UploadInvalidBodyStream (T.pack message))
        UploadStreamSizeMismatch ->
          throwOperationError err400 UploadSizeMismatch
        UploadStreamIdentityMismatch ->
          throwOperationError err400 UploadContentIdentityMismatch
    PutDispatchFailed (PutObjectBandwidthError bandwidthError) ->
      case bandwidthError of
        BandwidthUnavailable ->
          throwOperationError err503 UploadThrottleUnavailable
        BandwidthExceeded ->
          throwOperationError err429 UploadBandwidthExceeded
    PutDispatchFailed (PutObjectStoreError err) ->
      throwOperationError err409
        (UploadMutationRejected (renderS3Error err))

runDelete
  :: Env
  -> SessionUser
  -> Root
  -> RelativePath
  -> OperationId
  -> IO (Either Text MutationResult)
runDelete env user root rel op = do
  ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
  let uid = suUserId user
      key = buildObjectKey root uid rel
      pool = case root of
        Personal -> personalPool uid
        Shared   -> sharedPool
  admitted <- withTransaction (envDb env) $ \conn -> do
    existing <- mutationEpochExists
      conn uid Delete op (cfgIdempotencyWindowSec (envConfig env))
    available <-
      if existing then pure True else objectStoreAvailable (envS3 env)
    if available
      then Just <$> admitMutation conn uid Delete op pool key rel 0 Nothing
        (cfgMaxObjectBytes (envConfig env))
        (cfgIdempotencyWindowSec (envConfig env))
      else pure Nothing
  case admitted of
    Nothing -> pure (Left "object store unavailable")
    Just outcome ->
      case outcome of
        AdmitConflict -> pure (Right MutationConflict)
        AdmitRejected _ _ -> pure (Right MutationRejected)
        AdmitReplay MutationPending _ -> pure (Right MutationPending)
        AdmitReplay result _ -> pure (Right result)
        AdmitPrepared rec ->
          dispatchAndFinalize
            env
            user
            root
            rec
            "object request timed out"
            (deleteObject (envS3 env) key
              >>= pure . either (Left . renderS3Error) Right)

dispatchAndFinalize
  :: Env
  -> SessionUser
  -> Root
  -> MutationRecord
  -> failure
  -> IO (Either failure ())
  -> IO (Either failure MutationResult)
dispatchAndFinalize env user root rec timeoutFailure dispatch = do
  let cfg = envConfig env
      requestTimeoutSec = cfgObjectRequestTimeoutSec cfg
  prepared <- withTransaction (envDb env) $ \conn -> do
    moved <- markApplying conn rec
    if not moved
      then pure False
      else markExternalPending
        conn
        rec
        (requestTimeoutSec * 2)
        (cfgRecoveryGraceSec cfg)
  if not prepared
    then pure (Right MutationPending)
    else do
      dispatchOutcome <- timeout
        (fromInteger (requestTimeoutSec * 1000000))
        dispatch
      case dispatchOutcome of
        Nothing -> pure (Left timeoutFailure)
        Just (Left dispatchFailure) -> pure (Left dispatchFailure)
        Just (Right ()) -> do
          settled <- withTransaction (envDb env) $ \conn ->
            markExternalSettled conn rec
          if not settled
            then pure (Right MutationPending)
            else do
              observed <- observeObject (envS3 env) (mrObjectKey rec)
              finalized <- withTransaction (envDb env) $ \conn ->
                finalizeFromObservation conn rec observed
              case finalized of
                Just MutationSucceeded -> do
                  cacheInvalidatePrefixes (envRedis env) (suUserId user) root
                    (listCacheAncestorPrefixes (mrRelativePath rec))
                  pure (Right MutationSucceeded)
                Just MutationFailed -> pure (Right MutationFailed)
                Just other -> pure (Right other)
                Nothing -> pure (Right MutationPending)

observeObject :: S3 -> ObjectKey -> IO ObjectObservation
observeObject s3 key = do
  result <- observeObjectFully s3 key
  case result of
    Right info ->
      pure (ObjectPresent (oiSize info) (oiSha256 info))
    Left S3NotFound -> pure ObjectAbsent
    Left err -> pure (ObjectUnavailable (renderS3Error err))

listEntries
  :: Env
  -> SessionUser
  -> Root
  -> Text
  -> Maybe Text
  -> IO (Either Text ([ListEntry], Maybe Text))
listEntries env user root prefix continuation = do
  let cacheKey =
        "nuthatch:list:"
          <> unUserId (suUserId user)
          <> ":"
          <> rootLabel root
          <> ":"
          <> prefix
          <> ":"
          <> T.pack (show (cfgBrowsePageSize (envConfig env)))
          <> ":"
      s3Prefix =
        case root of
          Personal -> "users/" <> unUserId (suUserId user) <> "/" <> prefixSlash
          Shared   -> "shared/" <> prefixSlash
      prefixSlash
        | T.null prefix = ""
        | T.isSuffixOf "/" prefix = prefix
        | otherwise = prefix <> "/"
  cached <- case continuation of
    Nothing -> cacheGet (envRedis env) cacheKey
    Just _ -> pure Nothing
  case cached >>= A.decodeStrict of
    Just result -> pure (Right result)
    Nothing -> do
      page <- listObjects
        (envS3 env)
        s3Prefix
        (Just "/")
        continuation
        (cfgBrowsePageSize (envConfig env))
      case page of
        Left err -> pure (Left (renderS3Error err))
        Right lp -> do
          let rootPrefix =
                case root of
                  Personal -> "users/" <> unUserId (suUserId user) <> "/"
                  Shared   -> "shared/"
              folders =
                [ ListEntry
                    { leName = lastSeg (T.dropEnd 1 (T.drop (T.length rootPrefix) p))
                    , leIsPrefix = True
                    , leSize = Nothing
                    , leRelativePath = T.dropEnd 1 (T.drop (T.length rootPrefix) p)
                    }
                | p <- lpPrefixes lp
                , T.isPrefixOf rootPrefix p
                ]
              files =
                [ ListEntry
                    { leName = lastSeg rel
                    , leIsPrefix = False
                    , leSize = Just sz
                    , leRelativePath = rel
                    }
                | (k, sz) <- lpObjects lp
                , T.isPrefixOf rootPrefix k
                , let rel = T.drop (T.length rootPrefix) k
                , not (T.null rel)
                , not (T.isSuffixOf "/" rel)
                ]
              entries = folders <> files
              result = (entries, lpContinuationToken lp)
          when (continuation == Nothing) $
            cacheSet (envRedis env) cacheKey (cfgListCacheTtlSec (envConfig env))
              (BL.toStrict (A.encode result))
          pure (Right result)

searchEntries
  :: Env
  -> SessionUser
  -> Root
  -> Text
  -> IO (Either Text ([ListEntry], Bool))
searchEntries env user root q = do
  let s3Prefix =
        case root of
          Personal -> "users/" <> unUserId (suUserId user) <> "/"
          Shared   -> "shared/"
      needle = T.toCaseFold q
  started <- getMonotonicTimeNSec
  let deadline =
        started + fromInteger (cfgSearchTimeoutMs (envConfig env) * 1000000)
  keysResult <- collectKeysBounded
    (envS3 env)
    s3Prefix
    (cfgSearchMaxKeys (envConfig env))
    deadline
  pure $ case keysResult of
    Left err -> Left err
    Right (keys, partial) ->
      Right
        ( [ ListEntry
            { leName = lastSeg rel
            , leIsPrefix = False
            , leSize = Just sz
            , leRelativePath = rel
            }
        | (k, sz) <- keys
        , T.isPrefixOf s3Prefix k
        , let rel = T.drop (T.length s3Prefix) k
        , not (T.null rel)
        , needle `T.isInfixOf` T.toCaseFold rel
        ]
        , partial
        )

collectKeysBounded
  :: S3
  -> Text
  -> Int
  -> Word64
  -> IO (Either Text ([(Text, Integer)], Bool))
collectKeysBounded s3 prefix =
  collectKeysBoundedWith
    getMonotonicTimeNSec
    (\continuation limit ->
      first renderS3Error
        <$> listObjects s3 prefix Nothing continuation limit)

collectKeysBoundedWith
  :: IO Word64
  -> (Maybe Text -> Int -> IO (Either Text ListPage))
  -> Int
  -> Word64
  -> IO (Either Text ([(Text, Integer)], Bool))
collectKeysBoundedWith currentTime listPage maxKeys deadline =
  go Nothing [] 0
  where
    go mTok acc n
      | n >= maxKeys = pure (Right (take maxKeys acc, True))
      | otherwise = do
          now <- currentTime
          if now >= deadline
            then pure (Right (take maxKeys acc, True))
            else do
              let remainingMicros =
                    fromIntegral ((deadline - now + 999) `div` 1000)
              timedPage <- timeout remainingMicros $
                listPage mTok (min 1000 (maxKeys - n))
              case timedPage of
                Nothing -> pure (Right (take maxKeys acc, True))
                Just (Left err) -> pure (Left err)
                Just (Right lp) -> do
                  let acc' = acc <> lpObjects lp
                      n' = length acc'
                  if lpIsTruncated lp && n' < maxKeys
                    then go (lpContinuationToken lp) acc' n'
                    else pure
                      (Right (take maxKeys acc', lpIsTruncated lp))

runRecovery :: Env -> IO ()
runRecovery env = forever $ do
  recoverOnce env
  threadDelay (5 * 1000000)

recoverOnce :: Env -> IO ()
recoverOnce env = do
  let grace = cfgRecoveryGraceSec (envConfig env)
  batch <- withTransaction (envDb env) $ \conn -> claimActiveBatch conn 20 grace
  mapConcurrently_ (recoverOne env) batch

recoverOne :: Env -> MutationRecord -> IO ()
recoverOne env rec =
  case (mrState rec, mrExternal rec) of
    (Prepared, _) -> do
      settled <- withTransaction (envDb env) $ \conn -> do
        moved <- markApplying conn rec
        if moved then markSettledFromIdle conn rec else pure False
      when settled (finalizeRecovered env rec)
    (Applying, ExtIdle) -> do
      settled <- withTransaction (envDb env) $ \conn ->
        markSettledFromIdle conn rec
      when settled (finalizeRecovered env rec)
    (Applying, ExtPending) -> do
      settled <- withTransaction (envDb env) $ \conn ->
        markExternalSettled conn rec
      when settled (finalizeRecovered env rec)
    (Applying, ExtSettled) -> finalizeRecovered env rec
    _ -> pure ()

finalizeRecovered :: Env -> MutationRecord -> IO ()
finalizeRecovered env rec = do
  observed <- observeObjectWithRetry env rec
  result <- withTransaction (envDb env) $ \conn ->
    finalizeFromObservation conn rec observed
  when (result == Just MutationSucceeded) $
    cacheInvalidatePrefixes
      (envRedis env)
      (mrUserId rec)
      (recordRoot rec)
      (listCacheAncestorPrefixes (mrRelativePath rec))

observeObjectWithRetry :: Env -> MutationRecord -> IO ObjectObservation
observeObjectWithRetry env rec = do
  observed <- retrying
    recoveryRetryPolicy
    retryUnavailable
    (\_ -> observeObject (envS3 env) (mrObjectKey rec))
  case observed of
    ObjectUnavailable reason -> do
      audit "recovery_observation" "exhausted" (Just (unUserId (mrUserId rec)))
        [ ("operation_id", A.toJSON (unOperationId (mrOperationId rec)))
        , ("reason", A.toJSON reason)
        ]
      pure observed
    _ -> pure observed
  where
    retryUnavailable _ observation =
      case observation of
        ObjectUnavailable reason -> do
          audit "recovery_observation" "retry" (Just (unUserId (mrUserId rec)))
            [ ("operation_id", A.toJSON (unOperationId (mrOperationId rec)))
            , ("reason", A.toJSON reason)
            ]
          pure True
        _ -> pure False

recordRoot :: MutationRecord -> Root
recordRoot rec
  | mrPoolId rec == sharedPool = Shared
  | otherwise = Personal


lastSeg :: Text -> Text
lastSeg t =
  case reverse (T.splitOn "/" t) of
    (x : _) -> x
    []      -> t

requestCookies :: Request -> [(BS.ByteString, BS.ByteString)]
requestCookies request =
  maybe [] parseCookies (lookup hCookie (requestHeaders request))

parseCookieValueText
  :: A.ToJSON error
  => error
  -> BS.ByteString
  -> Handler Text
parseCookieValueText operationError bytes =
  case TE.decodeUtf8' bytes of
    Right value | not (T.null value) && T.length value <= 8192 -> pure value
    _ -> throwOperationError err400 operationError