1295 lines
43 KiB
Haskell
1295 lines
43 KiB
Haskell
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
|
|
( IOException
|
|
, SomeException
|
|
, fromException
|
|
, throwIO
|
|
, tryJust
|
|
)
|
|
import Control.Monad (forM, 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 (..), SqlError)
|
|
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, objectKeyFields)
|
|
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
|
|
, cacheGeneration
|
|
, cacheGet
|
|
, cacheInvalidatePrefixes
|
|
, cacheSet
|
|
, throttleCheck
|
|
, throttleTake
|
|
)
|
|
import Nuthatch.Recovery
|
|
( recoveryRetryPolicy
|
|
, trySynchronous
|
|
)
|
|
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 <- 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
|
|
|
|
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
|
|
|
|
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
|
|
|
|
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))
|
|
] <> objectKeyAudit (envConfig env) path)
|
|
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 "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 bodySource)
|
|
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))))
|
|
|
|
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)
|
|
}
|
|
|
|
handleDeletePrefixRequest
|
|
:: Env
|
|
-> SessionContext
|
|
-> DeletePrefixRequest
|
|
-> Handler DeletePrefixResponse
|
|
handleDeletePrefixRequest env session request = do
|
|
requireCsrf "delete_prefix" DeletePrefixCsrfRejected session (deletePrefixCsrf request)
|
|
liftIO $ audit "delete_prefix" "rejected"
|
|
(Just (unUserId (suUserId (sessionUser session))))
|
|
([ ("root", A.toJSON (rootLabel (deletePrefixRoot request)))
|
|
, ("reason", A.String "paged_delete_required")
|
|
] <> objectKeyAudit (envConfig env) (deletePrefixPath 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 "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)
|
|
|
|
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
|
|
|
|
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"
|
|
|
|
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")]
|
|
|
|
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"
|
|
|
|
objectKeyAudit :: Config -> RelativePath -> [(Text, A.Value)]
|
|
objectKeyAudit cfg =
|
|
objectKeyFields (cfgAuditObjectKeys cfg) . unRelativePath
|
|
|
|
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
|
|
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 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 && 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 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
|
|
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)
|
|
|
|
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
|
|
|
|
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
|
|
forM_ result $ \mutationResult ->
|
|
audit "recovery_mutation" (mutationAuditResult mutationResult)
|
|
(Just (unUserId (mrUserId rec)))
|
|
([ ("root", A.toJSON (rootLabel (recordRoot rec)))
|
|
, ("operation_id", A.toJSON (unOperationId (mrOperationId rec)))
|
|
] <> mutationReasonFields mutationResult
|
|
<> objectKeyAudit
|
|
(envConfig env)
|
|
(mrRelativePath rec))
|
|
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))
|