nuthatch/src/Nuthatch/App.hs

module Nuthatch.App
  ( Env (..)
  , newEnv
  , application
  , utf8Boundary
  , authenticationHandler
  , loginResponse
  , runRecovery
  ) where

import           Control.Concurrent (threadDelay)
import           Control.Exception (SomeException, try)
import           Control.Monad (forever, unless, void, when)
import           Control.Monad.IO.Class (liftIO)
import           Control.Monad.Trans.Except (runExceptT)
import           Crypto.Hash (Digest, SHA256, hashlazy)
import qualified Data.Aeson as A
import qualified Data.ByteArray as BA
import qualified Data.ByteString as BS
import qualified Data.ByteString.Base16 as B16
import qualified Data.ByteString.Lazy as BL
import           Data.Either (isRight)
import           Data.Maybe (fromMaybe, listToMaybe)
import           Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as TE
import           Data.UUID (toText)
import           Data.UUID.V4 (nextRandom)
import           Database.PostgreSQL.Simple (Only (..))
import           Network.HTTP.Types
  ( hContentType
  , hCookie
  , hLocation
  , status400
  , urlDecode
  )
import           Network.Wai (Request (..), 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, throttleTake)
import           Nuthatch.S3
  ( ListPage (..)
  , ObjectInfo (..)
  , S3
  , deleteObject
  , getObjectBytes
  , headObject
  , listObjects
  , putObject
  , renderS3Error
  )
import           Nuthatch.Session
import           Nuthatch.Types
import           Nuthatch.Ui (browsePage, browserScript)
import           Servant
import           Servant.Server.Experimental.Auth
  ( AuthHandler
  , mkAuthHandler
  )
import           Servant.Types.SourceT (runSourceT, source)
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)
  :<|> handleLoginRequest env
  :<|> handleOidcCallback env
  :<|> handleLogoutRequest env
  :<|> handleBrowseRequest env
  :<|> handleDownloadRequest env
  :<|> handleUploadRequest env
  :<|> handleDeleteRequest env
  :<|> handleDeletePrefixRequest env

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 "/" 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 -> listEntries env user root prefix
    Just query -> searchEntries env user root (unSearchQuery query)
  entries <- either
    (throwOperationError status502Error . BrowseObjectStoreFailed)
    pure
    result
  uploadOperation <- liftIO (toText <$> nextRandom)
  deleteOperations <- liftIO (mapM (\_ -> toText <$> nextRandom) entries)
  let response = BrowseResponse
        { browseEntries = entries
        , browseContinuation = Nothing
        , browsePartial = False
        }
      csrf = maybe "" unCsrfToken (sessionCsrf session)
  pure BrowseView
    { browseViewHtml = HtmlPage
        (browsePage
          user
          root
          prefix
          csrf
          uploadOperation
          deleteOperations
          entries
          Nothing
          "")
    , browseViewJson = response
    }

handleDownloadRequest
  :: Env
  -> SessionContext
  -> Root
  -> RelativePath
  -> Handler DownloadSuccess
handleDownloadRequest env session root path = do
  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
  throttle <- liftIO $
    throttleTake (envRedis env) (envConfig env) (suUserId user) (oiSize info)
  case throttle of
    Left _ -> throwOperationError err503 DownloadThrottleUnavailable
    Right False -> throwOperationError err429 DownloadBandwidthExceeded
    Right True -> pure ()
  result <- liftIO (getObjectBytes (envS3 env) key)
  body <- either
    (throwOperationError status502Error . DownloadObjectStoreFailed . renderS3Error)
    pure
    result
  liftIO $ audit "download" "ok" (Just (unUserId (suUserId user)))
    [ ("root", A.toJSON (rootLabel root))
    , ("bytes", A.toJSON (oiSize info))
    ]
  let filename = sanitizeFilename (basename path)
  pure $
    addHeader ("attachment; filename=\"" <> filename <> "\"") $
      addHeader "nosniff" (source (BL.toChunks 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
  chunksResult <- liftIO (runExceptT (runSourceT bodySource))
  chunks <- either
    (throwOperationError err400 . UploadInvalidBodyStream . T.pack)
    pure
    chunksResult
  let user = sessionUser session
      body = BL.fromChunks chunks
      size = unDeclaredSize declaredSize
      actualSize = toInteger (BL.length body)
      configuredMaximum = cfgMaxObjectBytes (envConfig env)
  unless (actualSize == size) $
    throwOperationError err400 UploadSizeMismatch
  when (size > configuredMaximum) $
    throwOperationError err413 UploadTooLarge
  unless (contentSha256 body == contentIdentity) $
    throwOperationError err400 UploadContentIdentityMismatch
  throttle <- liftIO $
    throttleTake (envRedis env) (envConfig env) (suUserId user) size
  case throttle of
    Left _ -> throwOperationError err503 UploadThrottleUnavailable
    Right False -> throwOperationError err429 UploadBandwidthExceeded
    Right True -> pure ()
  outcome <- liftIO (runPut env user root path operation size contentIdentity body)
  result <- either
    (throwOperationError err409 . UploadMutationRejected)
    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
    }

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
    }

handleDeletePrefixRequest
  :: Env
  -> SessionContext
  -> DeletePrefixRequest
  -> Handler DeletePrefixResponse
handleDeletePrefixRequest env session request = do
  requireCsrf DeletePrefixCsrfRejected session (deletePrefixCsrf request)
  result <- liftIO $
    deletePrefix
      env
      (sessionUser session)
      (deletePrefixRoot request)
      (unRelativePath (deletePrefixPath request))
  count <- either
    (throwOperationError err409 . DeletePrefixMutationRejected)
    pure
    result
  pure (DeletePrefixResponse count)

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"

sanitizeFilename :: Text -> Text
sanitizeFilename =
  T.map $ \character ->
    if character `elem` ("\"\\\r\n" :: String)
      then '_'
      else character

runPut
  :: Env
  -> SessionUser
  -> Root
  -> RelativePath
  -> OperationId
  -> Integer
  -> ContentSha256
  -> BL.ByteString
  -> IO (Either Text 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
  outcome <- withTransaction (envDb env) $ \conn ->
    admitMutation conn uid Put op pool key rel size (Just sha)
      (cfgMaxObjectBytes (envConfig env))
  case outcome of
    AdmitConflict -> pure (Right MutationConflict)
    AdmitRejected reason _ -> pure (Left reason)
    AdmitReplay MutationPending _ -> pure (Right MutationPending)
    AdmitReplay result _ -> pure (Right result)
    AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
      _ <- putObject (envS3 env) key size (Just sha) body
      pure ()

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
  outcome <- withTransaction (envDb env) $ \conn ->
    admitMutation conn uid Delete op pool key rel 0 Nothing
      (cfgMaxObjectBytes (envConfig env))
  case outcome of
    AdmitConflict -> pure (Right MutationConflict)
    AdmitRejected reason _ -> pure (Left reason)
    AdmitReplay MutationPending _ -> pure (Right MutationPending)
    AdmitReplay result _ -> pure (Right result)
    AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
      _ <- deleteObject (envS3 env) key
      pure ()

deletePrefix :: Env -> SessionUser -> Root -> Text -> IO (Either Text Int)
deletePrefix env user root prefix = do
  let listPrefix =
        case root of
          Personal ->
            "users/" <> unUserId (suUserId user) <> "/" <> withSlash prefix
          Shared ->
            "shared/" <> withSlash prefix
  keysResult <- collectKeys (envS3 env) listPrefix
  case keysResult of
    Left message -> pure (Left message)
    Right keys -> do
      let rootPrefix =
            case root of
              Personal -> "users/" <> unUserId (suUserId user) <> "/"
              Shared   -> "shared/"
          relativePaths =
            [ RelativePath relative
            | key <- keys
            , T.isPrefixOf rootPrefix key
            , let relative = T.drop (T.length rootPrefix) key
            , not (T.null relative)
            ]
      results <- mapM
        (\relativePath -> do
            operation <- OperationId . toText <$> nextRandom
            runDelete env user root relativePath operation)
        relativePaths
      let deleted = length [() | Right MutationSucceeded <- results]
          failure = listToMaybe [message | Left message <- results]
      pure (maybe (Right deleted) Left failure)
  where
    withSlash path
      | T.isSuffixOf "/" path = path
      | otherwise = path <> "/"

collectKeys :: S3 -> Text -> IO (Either Text [Text])
collectKeys s3 prefix = go Nothing []
  where
    go continuation accumulated = do
      page <- listObjects s3 prefix Nothing continuation 1000
      case page of
        Left message -> pure (Left (renderS3Error message))
        Right result -> do
          let accumulated' = accumulated <> map fst (lpObjects result)
          if lpIsTruncated result
            then go (lpContinuationToken result) accumulated'
            else pure (Right accumulated')

dispatchAndFinalize
  :: Env
  -> SessionUser
  -> Root
  -> MutationRecord
  -> IO ()
  -> IO (Either Text MutationResult)
dispatchAndFinalize env user root rec dispatch = do
  prepared <- withTransaction (envDb env) $ \conn -> do
    moved <- markApplying conn rec
    if not moved
      then pure False
      else markExternalPending conn rec
  if not prepared
    then pure (Right MutationPending)
    else do
      _ <- try dispatch :: IO (Either SomeException ())
      observed <- observeObject (envS3 env) (mrObjectKey rec)
      finalized <- withTransaction (envDb env) $ \conn -> do
        settled <- markExternalSettled conn rec
        if not settled
          then pure Nothing
          else 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 (Left "mutation failed after store settlement")
        Just other -> pure (Right other)
        Nothing -> pure (Right MutationPending)

observeObject :: S3 -> ObjectKey -> IO ObjectObservation
observeObject s3 key = do
  result <- headObject s3 key
  case result of
    Right (Just info@ObjectInfo { oiSha256 = Just _ }) ->
      pure (ObjectPresent (oiSize info) (oiSha256 info))
    Right (Just _) -> do
      body <- getObjectBytes s3 key
      pure $ case body of
        Right bytes ->
          ObjectPresent (toInteger (BL.length bytes)) (Just (contentSha256 bytes))
        Left err -> ObjectUnavailable (renderS3Error err)
    Right Nothing -> pure ObjectAbsent
    Left err -> pure (ObjectUnavailable (renderS3Error err))

listEntries :: Env -> SessionUser -> Root -> Text -> IO (Either Text [ListEntry])
listEntries env user root prefix = do
  let cacheKey =
        "nuthatch:list:"
          <> unUserId (suUserId user)
          <> ":"
          <> rootLabel root
          <> ":"
          <> prefix
          <> ":"
      s3Prefix =
        case root of
          Personal -> "users/" <> unUserId (suUserId user) <> "/" <> prefixSlash
          Shared   -> "shared/" <> prefixSlash
      prefixSlash
        | T.null prefix = ""
        | T.isSuffixOf "/" prefix = prefix
        | otherwise = prefix <> "/"
  cached <- cacheGet (envRedis env) cacheKey
  case cached >>= A.decodeStrict of
    Just entries -> pure (Right entries)
    Nothing -> do
      page <- listObjects (envS3 env) s3Prefix (Just "/") Nothing 1000
      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
          cacheSet (envRedis env) cacheKey (cfgListCacheTtlSec (envConfig env))
            (BL.toStrict (A.encode entries))
          pure (Right entries)

searchEntries :: Env -> SessionUser -> Root -> Text -> IO (Either Text [ListEntry])
searchEntries env user root q = do
  let s3Prefix =
        case root of
          Personal -> "users/" <> unUserId (suUserId user) <> "/"
          Shared   -> "shared/"
      needle = T.toLower q
  keysResult <- collectKeysBounded (envS3 env) s3Prefix (cfgSearchMaxKeys (envConfig env))
  pure $ case keysResult of
    Left err -> Left err
    Right keys ->
      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.toLower rel
        ]

collectKeysBounded :: S3 -> Text -> Int -> IO (Either Text [(Text, Integer)])
collectKeysBounded s3 prefix maxKeys = go Nothing [] 0
  where
    go mTok acc n
      | n >= maxKeys = pure (Right (take maxKeys acc))
      | otherwise = do
          page <- listObjects s3 prefix Nothing mTok (min 1000 (maxKeys - n))
          case page of
            Left err -> pure (Left (renderS3Error err))
            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'))

runRecovery :: Env -> IO ()
runRecovery env = forever $ do
  _ <- try (recoverOnce env) :: IO (Either SomeException ())
  threadDelay (5 * 1000000)

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

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

contentSha256 :: BL.ByteString -> ContentSha256
contentSha256 body =
  ContentSha256 . TE.decodeUtf8 . B16.encode . BA.convert $
    (hashlazy body :: Digest SHA256)

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