Split application orchestration by responsibility

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test check-dependencies check-fixture-lib check-kind-matrix check-storage-foundation check-auth check-app check-streaming check-idempotency check-recovery check-protocol check-product check-runtime check-v1 check-specs app-up app-down test-frontend check-frontend build-frontend
+.PHONY: test check-dependencies check-fixture-lib check-kind-matrix check-storage-foundation check-auth check-app check-app-boundaries check-streaming check-idempotency check-recovery check-protocol check-product check-runtime check-v1 check-specs app-up app-down test-frontend check-frontend build-frontend
 
 test:
 	./tests/run-host-tests.sh
@@ -21,6 +21,9 @@ check-auth:
 check-app:
 	./tests/kind/product-matrix-run.sh
 
+check-app-boundaries:
+	./tests/application-boundaries.sh
+
 check-streaming:
 	./tests/kind/streaming-run.sh
 

app/Main.hs

@@ -11,6 +11,7 @@ import           Network.Wai.Handler.Warp
 import           Network.Wai.Handler.WarpTLS (runTLS, tlsSettings)
 import           Nuthatch
 import           Nuthatch.Db (migrate, query_, withDb)
+import           Nuthatch.Mutation.Recovery (runRecovery)
 import           Nuthatch.Redis (withRedis)
 import           Nuthatch.Recovery (withLinkedWorker)
 import           Nuthatch.Runtime (retryStartup)

docs/spec/contracts.md

@@ -9,7 +9,7 @@ mutation protocol remains in [`Nuthatch.tla`](../protocol/Nuthatch.tla).
 
 | Contract | Implementation owner | Owning gate |
 | --- | --- | --- |
-| HTTP | `src/Nuthatch/Api.hs`, `src/Nuthatch/App.hs` | `make check-product` |
+| HTTP | `src/Nuthatch/Api.hs`, `src/Nuthatch/App.hs`, `src/Nuthatch/Http/Server.hs`, `src/Nuthatch/Http/Auth.hs`, `src/Nuthatch/Http/Error.hs`, `src/Nuthatch/Browse.hs`, `src/Nuthatch/Transfer.hs` | `make check-product` |
 | SQL | `sql/`, `src/Nuthatch/Db.hs`, `src/Nuthatch/Protocol.hs` | `make check-storage-foundation` |
 | Identity encoding | `src/Nuthatch/Paths.hs` | `make test` |
 | Content identity | `src/Nuthatch/Streaming.hs`, `src/Nuthatch/S3.hs` | `make check-streaming` |

nuthatch.cabal

@@ -100,9 +100,17 @@ library
     Nuthatch
     Nuthatch.Api
     Nuthatch.App
+    Nuthatch.App.Env
     Nuthatch.Audit
+    Nuthatch.Browse
     Nuthatch.Config
     Nuthatch.Db
+    Nuthatch.Http.Auth
+    Nuthatch.Http.Error
+    Nuthatch.Http.Server
+    Nuthatch.Mutation.Effects
+    Nuthatch.Mutation.Recovery
+    Nuthatch.Mutation.Service
     Nuthatch.Oidc
     Nuthatch.Paths
     Nuthatch.Protocol
@@ -112,6 +120,7 @@ library
     Nuthatch.S3
     Nuthatch.Session
     Nuthatch.Streaming
+    Nuthatch.Transfer
     Nuthatch.Types
 
 executable nuthatch

src/Nuthatch/App/Env.hs

@@ -0,0 +1,19 @@
+module Nuthatch.App.Env
+  ( Env (..)
+  , newEnv
+  ) where
+
+import           Nuthatch.Config (Config)
+import           Nuthatch.Db (Db)
+import           Nuthatch.Redis (Redis)
+import           Nuthatch.S3 (S3)
+
+data Env = Env
+  { envConfig :: Config
+  , envDb     :: Db
+  , envRedis  :: Redis
+  , envS3     :: S3
+  }
+
+newEnv :: Config -> Db -> Redis -> S3 -> Env
+newEnv = Env

src/Nuthatch/Browse.hs

@@ -0,0 +1,286 @@
+module Nuthatch.Browse
+  ( handleBrowseRequest
+  , handleDeletePlanRequest
+  , collectKeysBoundedWith
+  ) where
+
+import           Control.Monad (when)
+import           Control.Monad.IO.Class (liftIO)
+import qualified Data.Aeson as A
+import           Data.Bifunctor (first)
+import qualified Data.ByteString.Lazy as BL
+import           Data.Text (Text)
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import           Data.Word (Word64)
+import           GHC.Clock (getMonotonicTimeNSec)
+import           Nuthatch.Api
+import           Nuthatch.App.Env
+import           Nuthatch.Config
+import           Nuthatch.Db (ensurePersonalPool)
+import           Nuthatch.Http.Error
+import           Nuthatch.Redis
+  ( cacheGeneration
+  , cacheGet
+  , cacheSet
+  )
+import           Nuthatch.S3
+  ( ListPage (..)
+  , listObjects
+  , renderS3Error
+  )
+import           Nuthatch.Types
+import           Servant
+import           System.Timeout (timeout)
+
+handleBrowseRequest
+  :: Env
+  -> SessionContext
+  -> Maybe Root
+  -> Maybe BrowsePrefix
+  -> Maybe SearchQuery
+  -> Maybe PaginationToken
+  -> Handler BrowseResponse
+handleBrowseRequest env session selectedRoot selectedPrefix search continuation = do
+  let user = sessionUser session
+      root = maybe Personal id 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
+  pure BrowseResponse
+    { browseEntries = entries
+    , browseContinuation = PaginationToken <$> nextContinuation
+    , browsePartial = partial
+    }
+
+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
+    }
+
+listEntries
+  :: Env
+  -> SessionUser
+  -> Root
+  -> Text
+  -> Maybe Text
+  -> IO (Either Text ([ListEntry], Maybe Text))
+listEntries env user root prefix continuation = do
+  generation <- cacheGeneration (envRedis env) (suUserId user) root
+  let cacheOwner = case root of
+        Personal -> unUserId (suUserId user)
+        Shared   -> "shared"
+      cacheKey =
+        "nuthatch:list:"
+          <> cacheOwner
+          <> ":"
+          <> rootLabel root
+          <> ":"
+          <> prefix
+          <> ":"
+          <> T.pack (show (cfgBrowsePageSize (envConfig env)))
+          <> ":"
+          <> maybe "unavailable" TE.decodeUtf8 generation
+      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, generation) of
+    (Nothing, Just _) -> cacheGet (envRedis env) cacheKey
+    _ -> 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 listed -> 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) item))
+                    , leIsPrefix = True
+                    , leSize = Nothing
+                    , leRelativePath =
+                        T.dropEnd 1 (T.drop (T.length rootPrefix) item)
+                    }
+                | item <- lpPrefixes listed
+                , T.isPrefixOf rootPrefix item
+                ]
+              files =
+                [ ListEntry
+                    { leName = lastSeg relative
+                    , leIsPrefix = False
+                    , leSize = Just size
+                    , leRelativePath = relative
+                    }
+                | (key, size) <- lpObjects listed
+                , T.isPrefixOf rootPrefix key
+                , let relative = T.drop (T.length rootPrefix) key
+                , not (T.null relative)
+                , not (T.isSuffixOf "/" relative)
+                ]
+              entries = folders <> files
+              result = (entries, lpContinuationToken listed)
+          when (continuation == Nothing && generation /= 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 query = do
+  let s3Prefix =
+        case root of
+          Personal -> "users/" <> unUserId (suUserId user) <> "/"
+          Shared   -> "shared/"
+      needle = T.toCaseFold query
+  started <- getMonotonicTimeNSec
+  let deadline =
+        started + fromInteger (cfgSearchTimeoutMs (envConfig env) * 1000000)
+  keysResult <- collectKeysBounded
+    env
+    s3Prefix
+    (cfgSearchMaxKeys (envConfig env))
+    deadline
+  pure $ case keysResult of
+    Left err -> Left err
+    Right (keys, partial) ->
+      Right
+        ( [ ListEntry
+            { leName = lastSeg relative
+            , leIsPrefix = False
+            , leSize = Just size
+            , leRelativePath = relative
+            }
+        | (key, size) <- keys
+        , T.isPrefixOf s3Prefix key
+        , let relative = T.drop (T.length s3Prefix) key
+        , not (T.null relative)
+        , needle `T.isInfixOf` T.toCaseFold relative
+        ]
+        , partial
+        )
+
+collectKeysBounded
+  :: Env
+  -> Text
+  -> Int
+  -> Word64
+  -> IO (Either Text ([(Text, Integer)], Bool))
+collectKeysBounded env prefix =
+  collectKeysBoundedWith
+    getMonotonicTimeNSec
+    (\continuation limit ->
+      first renderS3Error
+        <$> listObjects (envS3 env) 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 token accumulated count
+      | count >= maxKeys =
+          pure (Right (take maxKeys accumulated, True))
+      | otherwise = do
+          now <- currentTime
+          if now >= deadline
+            then pure (Right (take maxKeys accumulated, True))
+            else do
+              let remainingMicros =
+                    fromIntegral ((deadline - now + 999) `div` 1000)
+              timedPage <- timeout remainingMicros $
+                listPage token (min 1000 (maxKeys - count))
+              case timedPage of
+                Nothing -> pure (Right (take maxKeys accumulated, True))
+                Just (Left err) -> pure (Left err)
+                Just (Right listed) -> do
+                  let accumulated' = accumulated <> lpObjects listed
+                      count' = length accumulated'
+                  if lpIsTruncated listed && count' < maxKeys
+                    then go
+                      (lpContinuationToken listed)
+                      accumulated'
+                      count'
+                    else pure
+                      (Right
+                        ( take maxKeys accumulated'
+                        , lpIsTruncated listed
+                        ))
+
+browsePrefixText :: BrowsePrefix -> Text
+browsePrefixText (BrowsePrefix Nothing) = ""
+browsePrefixText (BrowsePrefix (Just path)) = unRelativePath path
+
+lastSeg :: Text -> Text
+lastSeg text =
+  case reverse (T.splitOn "/" text) of
+    (segment : _) -> segment
+    [] -> text

src/Nuthatch/Http/Auth.hs

@@ -0,0 +1,234 @@
+module Nuthatch.Http.Auth
+  ( authenticationHandler
+  , handleLoginRequest
+  , handleOidcCallback
+  , handleLogoutRequest
+  , loginResponse
+  , requireCsrf
+  ) where
+
+import           Control.Monad.IO.Class (liftIO)
+import qualified Data.Aeson as A
+import qualified Data.ByteString as BS
+import           Data.Text (Text)
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import           Network.HTTP.Types
+  ( hCookie
+  , hLocation
+  )
+import           Network.Wai (Request (..))
+import           Nuthatch.Api
+import           Nuthatch.App.Env
+import           Nuthatch.Audit (audit)
+import           Nuthatch.Config
+import           Nuthatch.Db (ensurePersonalPool)
+import           Nuthatch.Http.Error
+import           Nuthatch.Oidc
+import           Nuthatch.Paths (encodeUserId)
+import           Nuthatch.Session
+import           Nuthatch.Types
+import           Servant
+import           Servant.Server.Experimental.Auth
+  ( AuthHandler
+  , mkAuthHandler
+  )
+import           Web.Cookie
+  ( defaultSetCookie
+  , parseCookies
+  , renderSetCookieBS
+  , sameSiteLax
+  , setCookieHttpOnly
+  , setCookieMaxAge
+  , setCookieName
+  , setCookiePath
+  , setCookieSameSite
+  , setCookieSecure
+  , setCookieValue
+  )
+
+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 <- case lookup sessionName cookies of
+    Nothing -> do
+      liftIO $ audit "authentication" "rejected" Nothing
+        [("reason", A.String "missing_session")]
+      throwError loginRequired
+    Just value -> pure value
+  sessionToken <- parseCookieValue validateSessionToken token
+  verified <- liftIO (verifySession cfg sessionToken)
+  user <- case verified of
+    Left _ -> do
+      liftIO $ audit "authentication" "rejected" Nothing
+        [("reason", A.String "invalid_session")]
+      throwError loginRequired
+    Right value -> pure value
+  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 _ -> rejectMalformed
+        Right value ->
+          either (const rejectMalformed) pure (validate value)
+    rejectMalformed = do
+      liftIO $ audit "authentication" "rejected" Nothing
+        [("reason", A.String "malformed_credential")]
+      throwOperationError err400 MalformedAuthentication
+
+handleLoginRequest :: Env -> Handler LoginStart
+handleLoginRequest env = do
+  let cfg = envConfig env
+  authorizationResult <- liftIO (beginAuthorization cfg)
+  authorization <- case authorizationResult of
+    Left _ -> do
+      liftIO $ audit "login_start" "rejected" Nothing
+        [("reason", A.String "provider_unavailable")]
+      throwOperationError err502 OidcAuthorizationFailed
+    Right value -> pure value
+  liftIO $ audit "login_start" "ok" Nothing []
+  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)
+  transactionBytes <- case lookup transactionName cookies of
+    Nothing -> rejectCallback "missing_transaction"
+    Just value -> pure value
+  transaction <- case TE.decodeUtf8' transactionBytes of
+    Right value | not (T.null value) && T.length value <= 8192 -> pure value
+    _ -> rejectCallback "malformed_transaction"
+  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
+    (\_ -> do
+      liftIO $ audit "login" "rejected" (Just (unUserId uid))
+        [("reason", A.String "session_creation_failed")]
+      throwOperationError err500
+        (SessionCreationFailed "session creation failed"))
+    pure
+    tokenResult
+  csrf <- liftIO newCsrfToken
+  liftIO $ audit "login" "ok" (Just (unUserId uid)) []
+  pure (loginResponse cfg token csrf)
+  where
+    rejectCallback reason = do
+      liftIO $ audit "login" "rejected" Nothing
+        [("reason", A.String reason)]
+      throwOperationError err400 OidcCallbackRejected
+
+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 "logout" LogoutCsrfRejected session providedCsrf
+  let cfg = envConfig env
+  liftIO $ audit "logout" "ok"
+    (Just (unUserId (suUserId (sessionUser session)))) []
+  pure $
+    addHeader (expiredCookieValue (cfgCookieName cfg) "/") $
+      addHeader (expiredCookieValue (cfgCsrfCookieName cfg) "/") $
+        addHeader
+          (expiredCookieValue
+            (cfgOidcTransactionCookieName cfg)
+            "/oidc/callback")
+          NoContent
+
+requireCsrf
+  :: A.ToJSON error
+  => Text
+  -> error
+  -> SessionContext
+  -> CsrfToken
+  -> Handler ()
+requireCsrf action csrfError session provided =
+  case sessionCsrf session of
+    Just expected | constantTimeEq expected provided -> pure ()
+    _ -> do
+      liftIO $ audit "authorization" "rejected"
+        (Just (unUserId (suUserId (sessionUser session))))
+        [ ("reason", A.String "csrf_rejected")
+        , ("operation", A.String action)
+        ]
+      throwOperationError err403 csrfError
+
+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
+        }
+
+requestCookies :: Request -> [(BS.ByteString, BS.ByteString)]
+requestCookies request =
+  maybe [] parseCookies (lookup hCookie (requestHeaders request))

src/Nuthatch/Http/Error.hs

@@ -0,0 +1,22 @@
+module Nuthatch.Http.Error
+  ( throwOperationError
+  , status502Error
+  ) where
+
+import qualified Data.Aeson as A
+import           Servant
+
+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 = []
+  }

src/Nuthatch/Http/Server.hs

@@ -0,0 +1,256 @@
+module Nuthatch.Http.Server
+  ( server
+  , utf8Boundary
+  ) where
+
+import           Control.Exception (SomeException)
+import           Control.Monad (forM, when)
+import           Control.Monad.IO.Class (liftIO)
+import qualified Data.Aeson as A
+import           Data.Either (isRight)
+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           Database.PostgreSQL.Simple (Only (..))
+import           Network.HTTP.Types
+  ( hContentType
+  , methodGet
+  , status200
+  , status400
+  , status404
+  , status405
+  , urlDecode
+  )
+import           Network.Wai (Request (..), responseFile, responseLBS)
+import           Nuthatch.Api
+import           Nuthatch.App.Env
+import           Nuthatch.Audit (audit)
+import           Nuthatch.Browse
+import           Nuthatch.Config
+import qualified Nuthatch.Db as Db
+import           Nuthatch.Http.Auth
+import           Nuthatch.Http.Error
+import           Nuthatch.Mutation.Effects
+import           Nuthatch.Mutation.Service
+import           Nuthatch.Recovery (trySynchronous)
+import           Nuthatch.Transfer
+import           Nuthatch.Types
+import           Servant
+import           System.Directory (doesFileExist)
+
+server :: Env -> Server Api
+server env =
+       pure (HealthResponse True)
+  :<|> handleReady env
+  :<|> handleAppRequest
+  :<|> staticAssetsServer
+  :<|> handleLoginRequest env
+  :<|> handleOidcCallback env
+  :<|> handleLogoutRequest env
+  :<|>
+    (    handleSessionRequest
+    :<|> handleBrowseRequest env
+    :<|> handleDownloadRequest env
+    :<|> handleUploadRequest env
+    :<|> handleDeleteRequest env
+    :<|> handleDeletePlanRequest env
+    :<|> handleDeleteBatchRequest 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'
+
+handleReady :: Env -> Handler ReadyResponse
+handleReady env = do
+  pg <- liftIO $
+    trySynchronous (Db.query_ (envDb env) "SELECT 1")
+      :: Handler (Either SomeException [Only Int])
+  case pg of
+    Right _ -> pure (ReadyResponse True)
+    Left _ -> throwOperationError err503 ReadyDependencyUnavailable
+
+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"
+
+handleDeleteRequest
+  :: Env
+  -> SessionContext
+  -> DeleteRequest
+  -> Handler DeleteResponse
+handleDeleteRequest env session request = do
+  requireCsrf "delete" DeleteCsrfRejected session (deleteCsrf request)
+  let user = sessionUser session
+      operation = deleteOperationId request
+      root = deleteRoot request
+  deleteAttempt <- liftIO $
+    tryRecoveryDependency
+      (runDelete env user root (deletePath request) operation)
+  outcome <- case deleteAttempt of
+    Left _ -> do
+      liftIO $ audit "delete" "failed"
+        (Just (unUserId (suUserId user)))
+        ([ ("root", A.toJSON (rootLabel root))
+         , ("operation_id", A.toJSON (unOperationId operation))
+         , ("reason", A.String "database_unavailable")
+         ] <> objectKeyAudit (envConfig env) (deletePath request))
+      throwOperationError err503
+        (DeleteMutationRejected "database unavailable")
+    Right value -> pure value
+  result <- case outcome of
+    Left reason -> do
+      liftIO $ audit "delete" "failed"
+        (Just (unUserId (suUserId user)))
+        ([ ("root", A.toJSON (rootLabel root))
+         , ("operation_id", A.toJSON (unOperationId operation))
+         , ("reason", A.String "object_store_unavailable")
+         ] <> objectKeyAudit (envConfig env) (deletePath request))
+      throwOperationError err409 (DeleteMutationRejected reason)
+    Right value -> pure value
+  liftIO $ audit "delete" (mutationAuditResult result)
+    (Just (unUserId (suUserId user)))
+    ([ ("root", A.toJSON (rootLabel root))
+     , ("operation_id", A.toJSON (unOperationId operation))
+     ] <> mutationReasonFields result
+       <> objectKeyAudit (envConfig env) (deletePath request))
+  pure DeleteResponse
+    { deleteResponseOperationId = operation
+    , deleteResponseResult = result
+    , deleteResponseIdempotencyWindowSeconds =
+        cfgIdempotencyWindowSec (envConfig env)
+    }
+
+handleDeleteBatchRequest
+  :: Env
+  -> SessionContext
+  -> DeleteBatchRequest
+  -> Handler DeleteBatchResponse
+handleDeleteBatchRequest env session request = do
+  requireCsrf
+    "delete_batch"
+    DeletePrefixCsrfRejected
+    session
+    (deleteBatchCsrf request)
+  let items = deleteBatchItems request
+  when (null items || length items > cfgDeletePageSize (envConfig env)) $ do
+    liftIO $ audit "delete_batch" "rejected"
+      (Just (unUserId (suUserId (sessionUser session))))
+      [ ("root", A.toJSON (rootLabel (deleteBatchRoot request)))
+      , ("reason", A.String "invalid_batch_size")
+      ]
+    throwOperationError err409
+      (DeletePrefixMutationRejected "delete batch size is invalid")
+  results <- forM items $ \item -> do
+    deleteAttempt <- liftIO $ tryRecoveryDependency $
+      runDelete
+        env
+        (sessionUser session)
+        (deleteBatchRoot request)
+        (deleteBatchPath item)
+        (deleteBatchOperationId item)
+    let outcome = case deleteAttempt of
+          Left _ -> Left "database unavailable"
+          Right value -> value
+        failureReason = case deleteAttempt of
+          Left _ -> "database_unavailable"
+          Right (Left _) -> "object_store_unavailable"
+          Right (Right _) -> ""
+    liftIO $ audit "delete" (either (const "failed") mutationAuditResult outcome)
+      (Just (unUserId (suUserId (sessionUser session))))
+      ([ ("root", A.toJSON (rootLabel (deleteBatchRoot request)))
+       , ("operation_id", A.toJSON
+          (unOperationId (deleteBatchOperationId item)))
+       ] <> either
+          (const [("reason", A.String failureReason)])
+          mutationReasonFields
+          outcome
+        <> objectKeyAudit (envConfig env) (deleteBatchPath 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)

src/Nuthatch/Mutation/Effects.hs

@@ -0,0 +1,50 @@
+module Nuthatch.Mutation.Effects
+  ( mutationAuditResult
+  , mutationReasonFields
+  , objectKeyAudit
+  , invalidateMutationCaches
+  ) where
+
+import qualified Data.Aeson as A
+import           Data.Text (Text)
+import           Nuthatch.App.Env
+import           Nuthatch.Audit (objectKeyFields)
+import           Nuthatch.Config
+import           Nuthatch.Paths (listCacheAncestorPrefixes)
+import           Nuthatch.Redis (cacheInvalidatePrefixes)
+import           Nuthatch.Types
+
+mutationAuditResult :: MutationResult -> Text
+mutationAuditResult MutationSucceeded = "ok"
+mutationAuditResult MutationPending   = "pending"
+mutationAuditResult MutationFailed    = "failed"
+mutationAuditResult MutationRejected  = "rejected"
+mutationAuditResult MutationConflict  = "conflict"
+
+mutationReasonFields :: MutationResult -> [(Text, A.Value)]
+mutationReasonFields MutationSucceeded = []
+mutationReasonFields MutationPending =
+  [("reason", A.String "settlement_pending")]
+mutationReasonFields MutationFailed =
+  [("reason", A.String "store_effect_failed")]
+mutationReasonFields MutationRejected =
+  [("reason", A.String "admission_rejected")]
+mutationReasonFields MutationConflict =
+  [("reason", A.String "operation_conflict")]
+
+objectKeyAudit :: Config -> RelativePath -> [(Text, A.Value)]
+objectKeyAudit cfg =
+  objectKeyFields (cfgAuditObjectKeys cfg) . unRelativePath
+
+invalidateMutationCaches
+  :: Env
+  -> UserId
+  -> Root
+  -> RelativePath
+  -> IO ()
+invalidateMutationCaches env user root path =
+  cacheInvalidatePrefixes
+    (envRedis env)
+    user
+    root
+    (listCacheAncestorPrefixes path)

src/Nuthatch/Mutation/Recovery.hs

@@ -0,0 +1,111 @@
+module Nuthatch.Mutation.Recovery
+  ( runRecovery
+  ) where
+
+import           Control.Concurrent (threadDelay)
+import           Control.Concurrent.Async (mapConcurrently_)
+import           Control.Monad (forM_, forever, when)
+import           Control.Retry (retrying)
+import qualified Data.Aeson as A
+import           Nuthatch.App.Env
+import           Nuthatch.Audit (audit)
+import           Nuthatch.Config
+import           Nuthatch.Db (withTransaction)
+import           Nuthatch.Mutation.Effects
+import           Nuthatch.Mutation.Service
+  ( observeObject
+  , tryRecoveryDependency
+  )
+import           Nuthatch.Protocol
+import           Nuthatch.Recovery (recoveryRetryPolicy)
+import           Nuthatch.Types
+
+runRecovery :: Env -> IO ()
+runRecovery env = forever $ do
+  recovered <- tryRecoveryDependency (recoverOnce env)
+  case recovered of
+    Left _ ->
+      audit "dependency_runtime" "unavailable" Nothing
+        [ ("dependency", A.String "postgresql")
+        , ("reason", A.String "recovery_cycle_failed")
+        ]
+    Right () -> pure ()
+  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 record =
+  case (mrState record, mrExternal record) of
+    (Prepared, _) -> do
+      settled <- withTransaction (envDb env) $ \conn -> do
+        moved <- markApplying conn record
+        if moved then markSettledFromIdle conn record else pure False
+      when settled (finalizeRecovered env record)
+    (Applying, ExtIdle) -> do
+      settled <- withTransaction (envDb env) $ \conn ->
+        markSettledFromIdle conn record
+      when settled (finalizeRecovered env record)
+    (Applying, ExtPending) -> do
+      settled <- withTransaction (envDb env) $ \conn ->
+        markExternalSettled conn record
+      when settled (finalizeRecovered env record)
+    (Applying, ExtSettled) -> finalizeRecovered env record
+    _ -> pure ()
+
+finalizeRecovered :: Env -> MutationRecord -> IO ()
+finalizeRecovered env record = do
+  observed <- observeObjectWithRetry env record
+  result <- withTransaction (envDb env) $ \conn ->
+    finalizeFromObservation conn record observed
+  forM_ result $ \mutationResult ->
+    audit "recovery_mutation" (mutationAuditResult mutationResult)
+      (Just (unUserId (mrUserId record)))
+      ([ ("root", A.toJSON (rootLabel (recordRoot record)))
+       , ("operation_id", A.toJSON (unOperationId (mrOperationId record)))
+       ] <> mutationReasonFields mutationResult
+         <> objectKeyAudit
+           (envConfig env)
+           (mrRelativePath record))
+  when (result == Just MutationSucceeded) $
+    invalidateMutationCaches
+      env
+      (mrUserId record)
+      (recordRoot record)
+      (mrRelativePath record)
+
+observeObjectWithRetry :: Env -> MutationRecord -> IO ObjectObservation
+observeObjectWithRetry env record = do
+  observed <- retrying
+    recoveryRetryPolicy
+    retryUnavailable
+    (\_ -> observeObject (envS3 env) (mrObjectKey record))
+  case observed of
+    ObjectUnavailable reason -> do
+      audit "recovery_observation" "exhausted"
+        (Just (unUserId (mrUserId record)))
+        [ ("operation_id", A.toJSON (unOperationId (mrOperationId record)))
+        , ("reason", A.toJSON reason)
+        ]
+      pure observed
+    _ -> pure observed
+  where
+    retryUnavailable _ observation =
+      case observation of
+        ObjectUnavailable reason -> do
+          audit "recovery_observation" "retry"
+            (Just (unUserId (mrUserId record)))
+            [ ("operation_id", A.toJSON (unOperationId (mrOperationId record)))
+            , ("reason", A.toJSON reason)
+            ]
+          pure True
+        _ -> pure False
+
+recordRoot :: MutationRecord -> Root
+recordRoot record
+  | mrPoolId record == sharedPool = Shared
+  | otherwise = Personal

src/Nuthatch/Mutation/Service.hs

@@ -0,0 +1,197 @@
+module Nuthatch.Mutation.Service
+  ( PutFailure (..)
+  , runPut
+  , runDelete
+  , dispatchAndFinalize
+  , observeObject
+  , tryRecoveryDependency
+  ) where
+
+import           Control.Exception
+  ( IOException
+  , SomeException
+  , fromException
+  , tryJust
+  )
+import           Data.Text (Text)
+import           Database.PostgreSQL.Simple (SqlError)
+import           Nuthatch.App.Env
+import           Nuthatch.Config
+import           Nuthatch.Db (ensurePersonalPool, withTransaction)
+import           Nuthatch.Mutation.Effects (invalidateMutationCaches)
+import           Nuthatch.Paths (buildObjectKey)
+import           Nuthatch.Protocol
+import           Nuthatch.S3
+  ( ObjectInfo (..)
+  , PutObjectError (..)
+  , S3
+  , S3Error (..)
+  , deleteObject
+  , objectStoreAvailable
+  , observeObjectFully
+  , renderS3Error
+  )
+import           Nuthatch.Types
+import           System.Timeout (timeout)
+
+runPut
+  :: Env
+  -> SessionUser
+  -> Root
+  -> RelativePath
+  -> OperationId
+  -> Integer
+  -> ContentSha256
+  -> (ObjectKey -> IO (Either PutFailure ()))
+  -> IO (Either PutFailure MutationResult)
+runPut env user root rel op size sha dispatchPut = 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 record ->
+          dispatchAndFinalize
+            env
+            user
+            root
+            record
+            (PutDispatchFailed (PutObjectStoreError S3TransportError))
+            (dispatchPut key)
+
+newtype PutFailure
+  = PutDispatchFailed PutObjectError
+  deriving stock (Eq, Show)
+
+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 record ->
+          dispatchAndFinalize
+            env
+            user
+            root
+            record
+            "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 record timeoutFailure dispatch = do
+  let cfg = envConfig env
+      requestTimeoutSec = cfgObjectRequestTimeoutSec cfg
+  prepared <- withTransaction (envDb env) $ \conn -> do
+    moved <- markApplying conn record
+    if not moved
+      then pure False
+      else markExternalPending
+        conn
+        record
+        (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 dispatchResult -> do
+          settled <- withTransaction (envDb env) $ \conn ->
+            markExternalSettled conn record
+          if not settled
+            then pure $ case dispatchResult of
+              Left dispatchFailure -> Left dispatchFailure
+              Right () -> Right MutationPending
+            else do
+              observed <- observeObject (envS3 env) (mrObjectKey record)
+              finalized <- withTransaction (envDb env) $ \conn ->
+                finalizeFromObservation conn record observed
+              case finalized of
+                Just MutationSucceeded -> do
+                  invalidateMutationCaches
+                    env
+                    (suUserId user)
+                    root
+                    (mrRelativePath record)
+                  pure (Right MutationSucceeded)
+                result -> pure $ case (dispatchResult, result) of
+                  (Left dispatchFailure, _) -> Left dispatchFailure
+                  (Right (), Just mutationResult) -> Right mutationResult
+                  (Right (), Nothing) -> 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))
+
+tryRecoveryDependency :: IO value -> IO (Either SomeException value)
+tryRecoveryDependency =
+  tryJust $ \exception ->
+    case ( fromException exception :: Maybe SqlError
+         , fromException exception :: Maybe IOException
+         ) of
+      (Just _, _) -> Just exception
+      (_, Just _) -> Just exception
+      _ -> Nothing

src/Nuthatch/Transfer.hs

@@ -0,0 +1,267 @@
+module Nuthatch.Transfer
+  ( handleDownloadRequest
+  , handleUploadRequest
+  , inspectBandwidth
+  , transferPieceSize
+  ) where
+
+import           Control.Exception (throwIO)
+import           Control.Monad (when)
+import           Control.Monad.IO.Class (liftIO)
+import qualified Data.Aeson as A
+import qualified Data.ByteString as BS
+import           Data.Text (Text)
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import qualified Network.HTTP.Types.URI as URI
+import           Nuthatch.Api
+import           Nuthatch.App.Env
+import           Nuthatch.Audit (audit)
+import           Nuthatch.Config
+import           Nuthatch.Http.Auth (requireCsrf)
+import           Nuthatch.Http.Error
+import           Nuthatch.Mutation.Effects
+import           Nuthatch.Mutation.Service
+import           Nuthatch.Paths (basename, buildObjectKey)
+import           Nuthatch.Redis
+  ( throttleCheck
+  , throttleTake
+  )
+import           Nuthatch.S3
+  ( ObjectInfo (..)
+  , PutObjectError (..)
+  , getObjectSource
+  , headObject
+  , putObjectStream
+  , renderS3Error
+  )
+import           Nuthatch.Streaming
+  ( BandwidthError (..)
+  , UploadStreamError (..)
+  )
+import           Nuthatch.Types
+import           Servant
+
+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))
+     ] <> objectKeyAudit (envConfig env) path)
+  pure $
+    addHeader (contentDisposition (basename path)) $
+      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 "put" UploadCsrfRejected session providedCsrf
+  let user = sessionUser session
+      size = unDeclaredSize declaredSize
+      configuredMaximum = cfgMaxObjectBytes (envConfig env)
+  when (size > configuredMaximum) $ do
+    liftIO $ audit "put" "rejected" (Just (unUserId (suUserId user)))
+      ([ ("root", A.toJSON (rootLabel root))
+       , ("bytes", A.toJSON size)
+       , ("operation_id", A.toJSON (unOperationId operation))
+       , ("reason", A.String "object_too_large")
+       ] <> objectKeyAudit (envConfig env) path)
+    throwOperationError err413 UploadTooLarge
+  requireUploadBandwidth env user root path operation size
+  putAttempt <- liftIO $
+    tryRecoveryDependency
+      (runPut
+        env
+        user
+        root
+        path
+        operation
+        size
+        contentIdentity
+        (\key ->
+          putObjectStream
+            (envS3 env)
+            key
+            size
+            (Just contentIdentity)
+            (transferPieceSize env)
+            (inspectBandwidth env user)
+            bodySource
+            >>= pure . either (Left . PutDispatchFailed) Right))
+  outcome <- case putAttempt of
+    Left _ -> do
+      liftIO $ audit "put" "failed" (Just (unUserId (suUserId user)))
+        ([ ("root", A.toJSON (rootLabel root))
+         , ("bytes", A.toJSON size)
+         , ("operation_id", A.toJSON (unOperationId operation))
+         , ("reason", A.String "database_unavailable")
+         ] <> objectKeyAudit (envConfig env) path)
+      throwOperationError err503
+        (UploadMutationRejected "database unavailable")
+    Right value -> pure value
+  result <- case outcome of
+    Left failure -> do
+      liftIO $ audit "put" "failed" (Just (unUserId (suUserId user)))
+        ([ ("root", A.toJSON (rootLabel root))
+         , ("bytes", A.toJSON size)
+         , ("operation_id", A.toJSON (unOperationId operation))
+         , ("reason", A.String (putFailureReason failure))
+         ] <> objectKeyAudit (envConfig env) path)
+      throwPutFailure failure
+    Right value -> pure value
+  liftIO $ audit "put" (mutationAuditResult result)
+    (Just (unUserId (suUserId user)))
+    ([ ("root", A.toJSON (rootLabel root))
+     , ("bytes", A.toJSON size)
+     , ("operation_id", A.toJSON (unOperationId operation))
+     ] <> mutationReasonFields result
+       <> objectKeyAudit (envConfig env) path)
+  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
+  -> Root
+  -> RelativePath
+  -> OperationId
+  -> Integer
+  -> Handler ()
+requireUploadBandwidth env user root path operation size = do
+  available <- liftIO $
+    throttleCheck
+      (envRedis env)
+      (envConfig env)
+      (suUserId user)
+      (min size (toInteger (transferPieceSize env)))
+  case available of
+    Left _ -> reject "throttle_unavailable" err503 UploadThrottleUnavailable
+    Right False ->
+      reject "bandwidth_exceeded" err429 UploadBandwidthExceeded
+    Right True -> pure ()
+  where
+    reject reason status response = do
+      liftIO $ audit "put" "rejected" (Just (unUserId (suUserId user)))
+        ([ ("root", A.toJSON (rootLabel root))
+         , ("bytes", A.toJSON size)
+         , ("operation_id", A.toJSON (unOperationId operation))
+         , ("reason", A.String reason)
+         ] <> objectKeyAudit (envConfig env) path)
+      throwOperationError status response
+
+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))))
+
+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))
+
+putFailureReason :: PutFailure -> Text
+putFailureReason (PutDispatchFailed (PutObjectStreamError streamError)) =
+  case streamError of
+    UploadStreamInvalid _ -> "invalid_body"
+    UploadStreamSizeMismatch -> "size_mismatch"
+    UploadStreamIdentityMismatch -> "content_identity_mismatch"
+putFailureReason (PutDispatchFailed (PutObjectBandwidthError bandwidthError)) =
+  case bandwidthError of
+    BandwidthUnavailable -> "throttle_unavailable"
+    BandwidthExceeded -> "bandwidth_exceeded"
+putFailureReason (PutDispatchFailed (PutObjectStoreError _)) =
+  "object_store_unavailable"
+
+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 = '_'

test/ProtocolConformance.hs

@@ -25,14 +25,47 @@ import           Nuthatch.App
 import           Nuthatch.Config
 import           Nuthatch.Db
 import qualified Nuthatch.Db as Db
+import qualified Nuthatch.Mutation.Service as Mutation
+import           Nuthatch.Mutation.Service (runDelete)
 import           Nuthatch.Protocol
 import           Nuthatch.Redis
 import           Nuthatch.S3
+import           Nuthatch.Transfer (inspectBandwidth, transferPieceSize)
 import           Nuthatch.Types
 import           Servant.Types.SourceT (SourceT (..), StepT (..), source)
 import           System.Environment (getArgs)
 import           System.IO (BufferMode (LineBuffering), hSetBuffering, stdout)
 
+runPut
+  :: Env
+  -> SessionUser
+  -> Root
+  -> RelativePath
+  -> OperationId
+  -> Integer
+  -> ContentSha256
+  -> SourceT IO BS.ByteString
+  -> IO (Either Mutation.PutFailure MutationResult)
+runPut env user root path operation size identity body =
+  Mutation.runPut
+    env
+    user
+    root
+    path
+    operation
+    size
+    identity
+    (\key ->
+      putObjectStream
+        (envS3 env)
+        key
+        size
+        (Just identity)
+        (transferPieceSize env)
+        (inspectBandwidth env user)
+        body
+        >>= pure . either (Left . Mutation.PutDispatchFailed) Right)
+
 main :: IO ()
 main = do
   hSetBuffering stdout LineBuffering

test/RecoveryPhase.hs

@@ -6,10 +6,11 @@ import qualified Data.ByteString as BS
 import qualified Data.ByteString.Lazy as BL
 import           Data.Text (Text)
 import qualified Data.Text as T
-import           Nuthatch.App (dispatchAndFinalize, newEnv)
+import           Nuthatch.App (newEnv)
 import           Nuthatch.Config
 import           Nuthatch.Db
 import qualified Nuthatch.Db as Db
+import           Nuthatch.Mutation.Service (dispatchAndFinalize)
 import           Nuthatch.Protocol
 import           Nuthatch.Redis
 import           Nuthatch.S3

test/Spec.hs

@@ -82,18 +82,18 @@ import           Network.Wai.Test
   )
 import           Nuthatch.Api
 import           Nuthatch.Audit (auditValue, objectKeyFields)
-import           Nuthatch.App
+import           Nuthatch.Browse (collectKeysBoundedWith)
-  ( authenticationHandler
-  , collectKeysBoundedWith
-  , loginResponse
-  , utf8Boundary
-  )
 import           Nuthatch.Config
   ( Config (..)
   , parseRedisUri
   , secret
   , validateAbsoluteUri
   )
+import           Nuthatch.Http.Auth
+  ( authenticationHandler
+  , loginResponse
+  )
+import           Nuthatch.Http.Server (utf8Boundary)
 import           Nuthatch.Oidc (OidcIdentity (..), verifyIdToken)
 import           Nuthatch.Paths
   ( buildObjectKey

tests/application-boundaries.sh

@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+for module in \
+  src/Nuthatch/Http/Server.hs \
+  src/Nuthatch/Http/Auth.hs \
+  src/Nuthatch/Browse.hs \
+  src/Nuthatch/Transfer.hs \
+  src/Nuthatch/Mutation/Service.hs \
+  src/Nuthatch/Mutation/Recovery.hs \
+  src/Nuthatch/Mutation/Effects.hs; do
+  test -f "$module"
+done
+
+test "$(wc -l <src/Nuthatch/App.hs)" -le 40
+
+if grep -Eq 'Servant|Nuthatch\.Http' \
+    src/Nuthatch/Mutation/Service.hs \
+    src/Nuthatch/Mutation/Recovery.hs \
+    src/Nuthatch/Mutation/Effects.hs; then
+  printf '%s\n' "Mutation protocol services depend on HTTP" >&2
+  exit 1
+fi
+
+grep -F 'Nuthatch.Mutation.Service' src/Nuthatch/Transfer.hs >/dev/null
+grep -F 'Nuthatch.Mutation.Service' src/Nuthatch/Http/Server.hs >/dev/null
+grep -F 'Nuthatch.Mutation.Service' src/Nuthatch/Mutation/Recovery.hs >/dev/null
+
+if grep -Eq 'recoverOnce|recoverOne|finalizeRecovered|observeObjectWithRetry' \
+    src/Nuthatch/App.hs src/Nuthatch/Http/*.hs; then
+  printf '%s\n' "Recovery implementation leaked into HTTP composition" >&2
+  exit 1
+fi
+
+printf '%s\n' "Application responsibility boundaries passed"

tests/spec-contracts.sh

@@ -18,7 +18,7 @@ for heading in \
 done
 
 for ownership in \
-  '| HTTP | `src/Nuthatch/Api.hs`, `src/Nuthatch/App.hs` | `make check-product` |' \
+  '| HTTP | `src/Nuthatch/Api.hs`, `src/Nuthatch/App.hs`, `src/Nuthatch/Http/Server.hs`, `src/Nuthatch/Http/Auth.hs`, `src/Nuthatch/Http/Error.hs`, `src/Nuthatch/Browse.hs`, `src/Nuthatch/Transfer.hs` | `make check-product` |' \
   '| SQL | `sql/`, `src/Nuthatch/Db.hs`, `src/Nuthatch/Protocol.hs` | `make check-storage-foundation` |' \
   '| Identity encoding | `src/Nuthatch/Paths.hs` | `make test` |' \
   '| Content identity | `src/Nuthatch/Streaming.hs`, `src/Nuthatch/S3.hs` | `make check-streaming` |' \
@@ -31,12 +31,17 @@ done
 for path in \
   src/Nuthatch/Api.hs \
   src/Nuthatch/App.hs \
+  src/Nuthatch/Browse.hs \
   src/Nuthatch/Config.hs \
   src/Nuthatch/Db.hs \
+  src/Nuthatch/Http/Auth.hs \
+  src/Nuthatch/Http/Error.hs \
+  src/Nuthatch/Http/Server.hs \
   src/Nuthatch/Paths.hs \
   src/Nuthatch/Protocol.hs \
   src/Nuthatch/S3.hs \
   src/Nuthatch/Streaming.hs \
+  src/Nuthatch/Transfer.hs \
   sql/001_initial_schema.sql \
   deploy/kubernetes/nuthatch.yaml.in \
   tests/kind/lib.sh \