Standalone.hs

{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE NoFieldSelectors #-}

-- | Open a store and serve its inspection surface without a Haskell host program.
module Kiroku.Metrics.Standalone
  ( InspectOptions (..),
    inspectOptionsParser,
    inspectParserInfo,
    InspectRuntime (..),
    resolveInspectOptions,
    InspectHooks (..),
    runInspect,
    InspectFailure (..),
    InspectBindError (..),
    classifyInspectFailure,
    renderInspectFailure,
    renderStartupBanner,
  )
where

import Control.Applicative (many, optional, (<|>))
import Control.Concurrent.STM (STM, TVar, atomically, newTVarIO, readTVar, writeTVar)
import Control.Exception (Exception, IOException, SomeException, catch, fromException, throwIO)
import Data.IORef (newIORef, readIORef, writeIORef)
import Data.IntMap.Strict qualified as IntMap
import Data.Text (Text)
import Data.Text qualified as T
import Hasql.Pool qualified as Pool
import Options.Applicative qualified as O

import Kiroku.Metrics.Capabilities
import Kiroku.Metrics.Collector
import Kiroku.Metrics.Config (defaultConfig)
import Kiroku.Metrics.Config qualified as Config
import Kiroku.Metrics.Cors
import Kiroku.Metrics.Health (postgresPing)
import Kiroku.Metrics.Server
import Kiroku.Store qualified as Store
import Kiroku.Store.Error (StoreError (..), mapGenericUsageError)
import Kiroku.Store.Notification (NotifierStartError (..))
import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition)

-- No Show instances: a database URL can contain credentials.
data InspectOptions = InspectOptions
  { databaseUrl :: !(Maybe Text),
    schema :: !(Maybe Text),
    poolSize :: !(Maybe Int),
    port :: !(Maybe Int),
    corsOrigins :: ![Text],
    corsAllowCredentials :: !(Maybe Bool),
    wsMaxConnections :: !(Maybe Int)
  }
  deriving stock (Eq)

data InspectRuntime = InspectRuntime
  { databaseUrl :: !Text,
    schema :: !Text,
    poolSize :: !Int,
    port :: !Int,
    cors :: !CorsPolicy,
    wsMaxConnections :: !Int
  }
  deriving stock (Eq)

data InspectHooks = InspectHooks
  { -- | Called once after successful binding, with the actual port and wiring.
    onListening :: !(Int -> Capabilities -> IO ()),
    waitForShutdown :: !(IO ())
  }

inspectOptionsParser :: O.Parser InspectOptions
inspectOptionsParser =
  InspectOptions
    <$> optional (textOption "database-url" "URL" "Database connection string (or DATABASE_URL)")
    <*> optional (textOption "schema" "NAME" "Migrated schema (default kiroku)")
    <*> optional (numberOption "pool-size" 1 intMaximum "Connection pool size (default 10)")
    <*> optional (numberOption "port" 0 65535 "Listen port (default 9091, all interfaces; 0 selects a loopback-only free port)")
    <*> many (textOption "cors-origin" "ORIGIN" "Allowed HTTP(S) browser origin; repeatable")
    <*> optional
      ( O.flag' True (O.long "cors-allow-credentials" <> O.help "Allow credentials for explicit origins")
          <|> O.flag' False (O.long "no-cors-allow-credentials" <> O.help "Disable credentials, overriding the environment")
      )
    <*> optional (numberOption "ws-max-connections" 1 intMaximum "Maximum WebSocket connections (default 100)")
  where
    textOption name metavar help = O.strOption (O.long name <> O.metavar metavar <> O.help help)
    numberOption name lower upper help =
      O.option
        (O.eitherReader (either (Left . T.unpack) Right . boundedDecimal lower upper . T.pack))
        (O.long name <> O.metavar "N" <> O.help help)

inspectParserInfo :: O.ParserInfo InspectOptions
inspectParserInfo =
  O.info
    (inspectOptionsParser O.<**> O.helper)
    ( O.failureCode 2
        <> O.fullDesc
        <> O.header "kiroku-inspect - standalone inspection server"
        <> O.footer "First SIGINT/SIGTERM shuts down gracefully; a second signal forces immediate exit 130."
        <> O.progDesc "Serve the read-only inspection API from a migrated Kiroku database. Runs no subscriptions."
    )

intMaximum :: Integer
intMaximum = toInteger (maxBound :: Int)

-- Parse into Integer and stop growing at the upper bound, before narrowing to Int.
boundedDecimal :: Integer -> Integer -> Text -> Either Text Int
boundedDecimal lower upper value
  | T.null value || not (T.all (\c -> c >= '0' && c <= '9') value) = Left "expected ASCII decimal digits"
  | otherwise = do
      n <- T.foldl' step (Right 0) value
      if n < lower then Left "value is below the allowed range" else Right (fromInteger n)
  where
    step result c = do
      n <- result
      let next = n * 10 + toInteger (fromEnum c - fromEnum '0')
      if next > upper then Left "value exceeds the allowed range" else Right next

resolveInspectOptions :: [(String, String)] -> InspectOptions -> Either Text InspectRuntime
resolveInspectOptions env opts = do
  url <- maybe (Left "kiroku-inspect: no database; pass --database-url or set DATABASE_URL (a libpq URI such as postgresql://user@host/db)") Right (opts.databaseUrl <|> variable "DATABASE_URL")
  nonempty "--database-url" url
  let schemaName = maybe "kiroku" id (opts.schema <|> variable "KIROKU_INSPECT_SCHEMA")
  nonempty "--schema" schemaName
  pool <- number opts.poolSize "KIROKU_INSPECT_POOL_SIZE" 1 intMaximum (Store.defaultConnectionSettings url).poolSize
  listenPort <- number opts.port "KIROKU_INSPECT_PORT" 0 65535 defaultConfig.port
  maxConnections <- number opts.wsMaxConnections "KIROKU_INSPECT_WS_MAX_CONNECTIONS" 1 intMaximum defaultConfig.wsMaxConnections
  credentials <- case opts.corsAllowCredentials of
    Just b -> Right b
    Nothing -> case variable "KIROKU_INSPECT_CORS_ALLOW_CREDENTIALS" of
      Nothing -> Right False
      Just "true" -> Right True
      Just "1" -> Right True
      Just "false" -> Right False
      Just "0" -> Right False
      Just _ -> Left "kiroku-inspect: KIROKU_INSPECT_CORS_ALLOW_CREDENTIALS must be true, false, 1 or 0"
  let originTexts = if null opts.corsOrigins then maybe [] (map T.strip . T.splitOn ",") (variable "KIROKU_INSPECT_CORS_ORIGINS") else opts.corsOrigins
  origins <-
    traverse
      ( \origin -> case allowedOrigin origin of
          Right o -> Right o
          Left WildcardOrigin -> Left "kiroku-inspect: wildcard '*' is not an allowed CORS origin"
          Left _ -> Left "kiroku-inspect: invalid CORS origin; use an explicit HTTP(S) origin"
      )
      originTexts
  let policy = if null origins then corsDisabled else (corsAllowOrigins origins) {allowCredentials = credentials}
  pure (InspectRuntime url schemaName pool listenPort policy maxConnections)
  where
    variable name = case T.pack <$> lookup name env of
      Just value | not (T.null value) -> Just value
      _ -> Nothing
    nonempty name value = if T.null (T.strip value) then Left ("kiroku-inspect: " <> name <> " must not be empty") else Right ()
    number explicit name lower upper fallback = case explicit of
      Just value | toInteger value >= lower && toInteger value <= upper -> Right value
      Just _ -> Left ("kiroku-inspect: " <> T.pack name <> " flag is outside the allowed range")
      Nothing -> case variable name of
        Nothing -> Right fallback
        Just value -> either (\message -> Left ("kiroku-inspect: " <> T.pack name <> ": " <> message)) Right (boundedDecimal lower upper value)

runInspect :: InspectHooks -> InspectRuntime -> IO ()
runInspect hooks rt = do
  storeVar <- newTVarIO Nothing
  metrics <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)
  let settings =
        (Store.defaultConnectionSettings rt.databaseUrl)
          { Store.schema = rt.schema,
            Store.poolSize = rt.poolSize,
            Store.eventHandler = Just (metricsEventHandler metrics Nothing),
            Store.observationHandler = Just (metricsObservationHandler metrics Nothing)
          }
      cfg = defaultConfig {Config.port = rt.port, Config.cors = rt.cors, Config.wsMaxConnections = rt.wsMaxConnections}
  Store.withStore settings $ \store -> do
    atomically (writeTVar storeVar (Just store))
    providers <- storeServerProviders cfg metrics store
    listening <- newIORef False
    let serve = withMetricsServerWithProviders cfg metrics [postgresPing store] providers $ \server -> do
          writeIORef listening True
          hooks.onListening server.serverPort (capabilitiesFor cfg (providerPresence providers))
          hooks.waitForShutdown
    serve `catch` \err -> do
      started <- readIORef listening
      if started then throwIO (err :: IOException) else throwIO (InspectBindError err)

-- | Only listener setup IO failures carry this marker. Store acquisition and
-- user hooks remain outside that boundary; asynchronous exceptions pass through.
newtype InspectBindError = InspectBindError IOException
  deriving stock (Show)
  deriving anyclass (Exception)

-- | Sanitized diagnostic classes, independent of driver messages and SQL text.
data InspectFailure = BindFailure | DatabaseFailure | SchemaFailure | OtherFailure
  deriving stock (Eq, Show)

classifyInspectFailure :: SomeException -> InspectFailure
classifyInspectFailure err
  | Just (_ :: InspectBindError) <- fromException err = BindFailure
  | Just notifier <- fromException err = case notifier of
      NotifierConnectError _ -> DatabaseFailure
      NotifierListenError nested -> case fromException nested :: Maybe Pool.UsageError of
        Just usage -> classifyStoreFailure (mapGenericUsageError usage)
        Nothing -> maybe DatabaseFailure classifyStoreFailure (fromException nested)
  | Just usage <- fromException err = classifyStoreFailure (mapGenericUsageError usage)
  | Just storeError <- fromException err = classifyStoreFailure storeError
  | otherwise = OtherFailure

classifyStoreFailure :: StoreError -> InspectFailure
classifyStoreFailure = \case
  UnexpectedServerError code _ | code == "42P01" || code == "3F000" -> SchemaFailure
  UnexpectedServerError _ _ -> DatabaseFailure
  ConnectionError _ -> DatabaseFailure
  ConnectionLost _ -> DatabaseFailure
  PoolAcquisitionTimeout -> DatabaseFailure
  TransientTransactionFailure _ _ -> DatabaseFailure
  _ -> OtherFailure

renderInspectFailure :: InspectFailure -> Text
renderInspectFailure failure = "kiroku-inspect: " <> label <> " failure (details redacted)"
  where
    label = case failure of
      BindFailure -> "bind"
      DatabaseFailure -> "database"
      SchemaFailure -> "schema"
      OtherFailure -> "other"

readPosition :: TVar (Maybe Store.KirokuStore) -> STM Store.GlobalPosition
readPosition storeVar = readTVar storeVar >>= maybe (pure (Store.GlobalPosition 0)) (publisherPosition . (.publisher))

readSubscribers :: TVar (Maybe Store.KirokuStore) -> STM Int
readSubscribers storeVar = readTVar storeVar >>= maybe (pure 0) (\store -> IntMap.size <$> readTVar (subscribers store.publisher))

-- | Never includes the database URL. The port is the actual bound port.
renderStartupBanner :: InspectRuntime -> Int -> Capabilities -> [Text]
renderStartupBanner rt boundPort caps =
  [ "kiroku-inspect: connected to schema " <> T.pack (show rt.schema) <> "; listening on port " <> T.pack (show boundPort),
    "kiroku-inspect: routes browse="
      <> enabled caps.routes.browse
      <> " subscriptions_checkpoints="
      <> enabled caps.routes.subscriptionsCheckpoints
      <> " dead_letters="
      <> enabled caps.routes.deadLetters
      <> " subscriptions_live="
      <> enabled caps.routes.subscriptionsLive
      <> " websocket_events="
      <> enabled caps.routes.websocketEvents
      <> " cors="
      <> enabled caps.corsIsEnabled,
    "kiroku-inspect: this process runs no subscriptions; /subscriptions, /metrics, and /health reflect only this process"
  ]
  where
    enabled True = "on"
    enabled False = "off"