Checkpoints.hs

-- | Read-only durable checkpoint inventory, independent of the live registry.
module Kiroku.Metrics.Checkpoints
  ( CheckpointInventoryProvider,
    storeCheckpointInventory,
    CheckpointInventoryResponse (..),
    CheckpointRow (..),
    checkpointInventoryResponse,
    checkpointsPath,
    checkpointsApp,
    checkpointsNotConfiguredApp,
  )
where

import Data.Aeson (FromJSON (..), ToJSON (..), encode, object, withObject, (.:), (.=))
import Data.Int (Int32, Int64)
import Data.Text (Text)
import Data.Time (UTCTime)
import Data.Vector qualified as V
import Network.HTTP.Types (status200, status404, status405)
import Network.Wai (Application, Response, mapResponseHeaders, pathInfo, requestMethod, responseHeaders, responseLBS, responseStatus)

import Kiroku.Metrics.JSON (errorResponse, jsonResponse, storeErrorResponse)
import Kiroku.Store (GlobalPosition (..), KirokuStore, StoreError, SubscriptionCheckpoint (..), SubscriptionCheckpointInventory (..), SubscriptionName (..), runStoreIO, subscriptionCheckpointInventory)

-- | One call reads the frontier and all checkpoint rows in one SQL snapshot.
-- Inventory work is proportional to row count; clients should not overlap polls.
type CheckpointInventoryProvider = IO (Either StoreError SubscriptionCheckpointInventory)

storeCheckpointInventory :: KirokuStore -> CheckpointInventoryProvider
storeCheckpointInventory store = runStoreIO store subscriptionCheckpointInventory

data CheckpointRow = CheckpointRow
  { subscription :: !Text,
    member :: !Int32,
    checkpointPosition :: !Int64,
    updatedAt :: !UTCTime
  }
  deriving stock (Eq, Show)

data CheckpointInventoryResponse = CheckpointInventoryResponse
  { storePosition :: !Int64,
    checkpoints :: ![CheckpointRow]
  }
  deriving stock (Eq, Show)

instance ToJSON CheckpointRow where
  toJSON row = object ["subscription" .= row.subscription, "member" .= row.member, "checkpoint_position" .= row.checkpointPosition, "updated_at" .= row.updatedAt]

instance FromJSON CheckpointRow where
  parseJSON = withObject "CheckpointRow" $ \o -> CheckpointRow <$> o .: "subscription" <*> o .: "member" <*> o .: "checkpoint_position" <*> o .: "updated_at"

instance ToJSON CheckpointInventoryResponse where
  toJSON inventory = object ["store_position" .= inventory.storePosition, "checkpoints" .= inventory.checkpoints]

instance FromJSON CheckpointInventoryResponse where
  parseJSON = withObject "CheckpointInventoryResponse" $ \o -> CheckpointInventoryResponse <$> o .: "store_position" <*> o .: "checkpoints"

checkpointInventoryResponse :: SubscriptionCheckpointInventory -> CheckpointInventoryResponse
checkpointInventoryResponse (SubscriptionCheckpointInventory (GlobalPosition position) rows) =
  CheckpointInventoryResponse position (map toRow (V.toList rows))
  where
    toRow (SubscriptionCheckpoint (SubscriptionName name) index (GlobalPosition cp) updated) = CheckpointRow name index cp updated

checkpointsPath :: [Text]
checkpointsPath = ["subscription-checkpoints"]

-- | GET and HEAD only. Unknown query parameters are ignored. Expected store
-- failures are sanitized; thrown exceptions (including cancellation) propagate.
checkpointsApp :: CheckpointInventoryProvider -> Application
checkpointsApp provider = checkpointsResponseApp $ either (storeErrorResponse "checkpoint_inventory_unavailable") (jsonResponse status200 . encode . checkpointInventoryResponse) <$> provider

-- | Same method and HEAD behavior when a server has no inventory provider.
checkpointsNotConfiguredApp :: Application
checkpointsNotConfiguredApp = checkpointsResponseApp $ pure $ errorResponse status404 "checkpoint_inventory_not_configured" "Durable checkpoint inventory is not configured." Nothing

checkpointsResponseApp :: IO Response -> Application
checkpointsResponseApp readResponse req respond = do
  response <-
    if pathInfo req /= checkpointsPath
      then pure $ errorResponse status404 "not_found" "Not found" Nothing
      else
        if requestMethod req `notElem` ["GET", "HEAD"]
          then pure $ mapResponseHeaders (("Allow", "GET, HEAD") :) $ errorResponse status405 "method_not_allowed" "Use GET or HEAD." Nothing
          else readResponse
  respond $
    if requestMethod req == "HEAD"
      then responseLBS (responseStatus response) (responseHeaders response) ""
      else response