Types.hs

{-# LANGUAGE OverloadedStrings #-}

module JetStream.Types
  ( AccountAPIStats (..)
  , AccountInfo (..)
  , AccountLimits (..)
  , AccountTier (..)
  , StreamName
  , ConsumerName
  , Subject
  , Payload
  , Headers
  , CallOption
  , JetStreamRequestOption
  , applyRequestOptions
  , withRequestTimeout
  , AckPolicy (..)
  , DeliverPolicy (..)
  , DiscardPolicy (..)
  , ReplayPolicy (..)
  , RetentionPolicy (..)
  , StorageType (..)
  , byteStringToJSON
  , parseByteString
  , diffTimeNanosToJSON
  , applyCallOptions
  ) where

import           Data.Aeson
import qualified Data.Aeson.Key     as Key
import qualified Data.Aeson.KeyMap  as KeyMap
import           Data.Aeson.Types   (Pair, Parser, typeMismatch)
import qualified Data.ByteString    as BS
import           Data.Maybe         (catMaybes, fromMaybe)
import qualified Data.Text          as T
import qualified Data.Text.Encoding as E
import           Data.Time.Clock    (NominalDiffTime)
import qualified Data.Time.Clock    as Time
import           Data.Word          (Word64)

type StreamName = BS.ByteString
type ConsumerName = BS.ByteString
type Subject = BS.ByteString
type Payload = BS.ByteString
type Headers = [(BS.ByteString, BS.ByteString)]
type CallOption a = a -> a

-- | Options scoped to one JetStream API request. This is distinct from
-- stream, consumer, publish, and fetch configuration.
newtype JetStreamRequestOption = JetStreamRequestOption (JetStreamRequestConfig -> JetStreamRequestConfig)

newtype JetStreamRequestConfig = JetStreamRequestConfig { requestConfigTimeoutMicros :: Maybe Int }

applyRequestOptions :: Int -> [JetStreamRequestOption] -> Int
applyRequestOptions defaultTimeout options =
  fromMaybe defaultTimeout
    (requestConfigTimeoutMicros (foldl apply emptyRequestConfig options))
  where
    emptyRequestConfig = JetStreamRequestConfig Nothing
    apply value (JetStreamRequestOption option) = option value

-- | Override the timeout for one JetStream request.
withRequestTimeout :: NominalDiffTime -> JetStreamRequestOption
withRequestTimeout timeoutSeconds =
  JetStreamRequestOption $ \config ->
    config
      { requestConfigTimeoutMicros =
          Just (max 1 (floor (realToFrac timeoutSeconds * (1000000 :: Double))))
      }

data AccountInfo = AccountInfo
                     { accountInfoTier   :: AccountTier
                     , accountInfoDomain :: Maybe BS.ByteString
                     , accountInfoAPI    :: AccountAPIStats
                     , accountInfoTiers  :: [(BS.ByteString, AccountTier)]
                     }
  deriving (Eq, Show)

data AccountTier = AccountTier
                     { accountTierMemory          :: Integer
                     , accountTierStorage         :: Integer
                     , accountTierReservedMemory  :: Integer
                     , accountTierReservedStorage :: Integer
                     , accountTierStreams         :: Int
                     , accountTierConsumers       :: Int
                     , accountTierLimits          :: AccountLimits
                     }
  deriving (Eq, Show)

data AccountAPIStats = AccountAPIStats
                         { accountAPILevel    :: Int
                         , accountAPITotal    :: Integer
                         , accountAPIErrors   :: Integer
                         , accountAPIInflight :: Maybe Integer
                         }
  deriving (Eq, Show)

data AccountLimits = AccountLimits
                       { accountLimitsMaxMemory             :: Integer
                       , accountLimitsMaxStorage            :: Integer
                       , accountLimitsMaxStreams            :: Int
                       , accountLimitsMaxConsumers          :: Int
                       , accountLimitsMaxAckPending         :: Int
                       , accountLimitsMemoryMaxStreamBytes  :: Integer
                       , accountLimitsStorageMaxStreamBytes :: Integer
                       , accountLimitsMaxBytesRequired      :: Bool
                       }
  deriving (Eq, Show)

applyCallOptions :: [CallOption a] -> a -> a
applyCallOptions options value =
  foldl (flip ($)) value options

data AckPolicy = AckNone
               | AckAll
               | AckExplicit
               | AckPolicyUnknown T.Text
  deriving (Eq, Show)

data DeliverPolicy = DeliverAll
                   | DeliverLast
                   | DeliverNew
                   | DeliverByStartSequence Word64
                   | DeliverByStartTime Time.UTCTime
                   | DeliverLastPerSubject
                   | DeliverPolicyUnknown T.Text
  deriving (Eq, Show)

data DiscardPolicy = DiscardOld
                   | DiscardNew
                   | DiscardPolicyUnknown T.Text
  deriving (Eq, Show)

data ReplayPolicy = ReplayInstant
                  | ReplayOriginal
                  | ReplayPolicyUnknown T.Text
  deriving (Eq, Show)

data RetentionPolicy = LimitsPolicy
                     | InterestPolicy
                     | WorkQueuePolicy
                     | RetentionPolicyUnknown T.Text
  deriving (Eq, Show)

data StorageType = FileStorage
                 | MemoryStorage
                 | StorageTypeUnknown T.Text
  deriving (Eq, Show)

instance ToJSON AckPolicy where
  toJSON policy =
    String $
      case policy of
        AckNone                -> "none"
        AckAll                 -> "all"
        AckExplicit            -> "explicit"
        AckPolicyUnknown value -> value

instance FromJSON AckPolicy where
  parseJSON = withText "AckPolicy" $ \value ->
    case value of
      "none"     -> pure AckNone
      "all"      -> pure AckAll
      "explicit" -> pure AckExplicit
      _          -> pure (AckPolicyUnknown value)

instance ToJSON DeliverPolicy where
  toJSON policy =
    String $
      case policy of
        DeliverAll                 -> "all"
        DeliverLast                -> "last"
        DeliverNew                 -> "new"
        DeliverByStartSequence _   -> "by_start_sequence"
        DeliverByStartTime _       -> "by_start_time"
        DeliverLastPerSubject      -> "last_per_subject"
        DeliverPolicyUnknown value -> value

instance FromJSON DeliverPolicy where
  parseJSON = withText "DeliverPolicy" $ \value ->
    case value of
      "all"               -> pure DeliverAll
      "last"              -> pure DeliverLast
      "new"               -> pure DeliverNew
      "by_start_sequence" -> fail "deliver policy by_start_sequence requires opt_start_seq"
      "by_start_time"     -> fail "deliver policy by_start_time requires opt_start_time"
      "last_per_subject"  -> pure DeliverLastPerSubject
      _                   -> pure (DeliverPolicyUnknown value)

instance ToJSON DiscardPolicy where
  toJSON policy =
    String $
      case policy of
        DiscardOld                 -> "old"
        DiscardNew                 -> "new"
        DiscardPolicyUnknown value -> value

instance FromJSON DiscardPolicy where
  parseJSON = withText "DiscardPolicy" $ \value ->
    case value of
      "old" -> pure DiscardOld
      "new" -> pure DiscardNew
      _     -> pure (DiscardPolicyUnknown value)

instance ToJSON ReplayPolicy where
  toJSON policy =
    String $
      case policy of
        ReplayInstant             -> "instant"
        ReplayOriginal            -> "original"
        ReplayPolicyUnknown value -> value

instance FromJSON ReplayPolicy where
  parseJSON = withText "ReplayPolicy" $ \value ->
    case value of
      "instant"  -> pure ReplayInstant
      "original" -> pure ReplayOriginal
      _          -> pure (ReplayPolicyUnknown value)

instance ToJSON RetentionPolicy where
  toJSON policy =
    String $
      case policy of
        LimitsPolicy                 -> "limits"
        InterestPolicy               -> "interest"
        WorkQueuePolicy              -> "workqueue"
        RetentionPolicyUnknown value -> value

instance FromJSON RetentionPolicy where
  parseJSON = withText "RetentionPolicy" $ \value ->
    case value of
      "limits"    -> pure LimitsPolicy
      "interest"  -> pure InterestPolicy
      "workqueue" -> pure WorkQueuePolicy
      _           -> pure (RetentionPolicyUnknown value)

instance ToJSON StorageType where
  toJSON storage =
    String $
      case storage of
        FileStorage              -> "file"
        MemoryStorage            -> "memory"
        StorageTypeUnknown value -> value

instance FromJSON StorageType where
  parseJSON = withText "StorageType" $ \value ->
    case value of
      "file"   -> pure FileStorage
      "memory" -> pure MemoryStorage
      _        -> pure (StorageTypeUnknown value)

byteStringToJSON :: BS.ByteString -> Value
byteStringToJSON =
  String . E.decodeUtf8

parseByteString :: Value -> Parser BS.ByteString
parseByteString =
  withText "ByteString" (pure . E.encodeUtf8)

diffTimeNanosToJSON :: NominalDiffTime -> Value
diffTimeNanosToJSON diffTime =
  Number (fromInteger (floor (realToFrac diffTime * (1000000000 :: Double))))

instance FromJSON AccountInfo where
  parseJSON value@(Object obj) =
    AccountInfo
      <$> parseJSON value
      <*> parseOptionalByteStringField obj "domain"
      <*> obj .: "api"
      <*> parseAccountTiers obj
  parseJSON invalid =
    typeMismatch "AccountInfo" invalid

instance ToJSON AccountInfo where
  toJSON info =
    object $
      accountTierPairs (accountInfoTier info)
        ++ catMaybes
          [ maybeByteStringPair "domain" (accountInfoDomain info)
          , Just ("api" .= accountInfoAPI info)
          , Just ("tiers" .= accountTiersValue (accountInfoTiers info))
          ]

instance FromJSON AccountTier where
  parseJSON =
    withObject "AccountTier" $ \obj ->
      AccountTier
        <$> obj .: "memory"
        <*> obj .: "storage"
        <*> obj .: "reserved_memory"
        <*> obj .: "reserved_storage"
        <*> obj .: "streams"
        <*> obj .: "consumers"
        <*> obj .: "limits"

instance ToJSON AccountTier where
  toJSON =
    object . accountTierPairs

instance FromJSON AccountAPIStats where
  parseJSON =
    withObject "AccountAPIStats" $ \obj ->
      AccountAPIStats
        <$> obj .: "level"
        <*> obj .: "total"
        <*> obj .: "errors"
        <*> obj .:? "inflight"

instance ToJSON AccountAPIStats where
  toJSON stats =
    object . catMaybes $
      [ Just ("level" .= accountAPILevel stats)
      , Just ("total" .= accountAPITotal stats)
      , Just ("errors" .= accountAPIErrors stats)
      , maybePair "inflight" (accountAPIInflight stats)
      ]

instance FromJSON AccountLimits where
  parseJSON =
    withObject "AccountLimits" $ \obj ->
      AccountLimits
        <$> obj .: "max_memory"
        <*> obj .: "max_storage"
        <*> obj .: "max_streams"
        <*> obj .: "max_consumers"
        <*> obj .: "max_ack_pending"
        <*> obj .: "memory_max_stream_bytes"
        <*> obj .: "storage_max_stream_bytes"
        <*> obj .: "max_bytes_required"

instance ToJSON AccountLimits where
  toJSON limits =
    object
      [ "max_memory" .= accountLimitsMaxMemory limits
      , "max_storage" .= accountLimitsMaxStorage limits
      , "max_streams" .= accountLimitsMaxStreams limits
      , "max_consumers" .= accountLimitsMaxConsumers limits
      , "max_ack_pending" .= accountLimitsMaxAckPending limits
      , "memory_max_stream_bytes" .= accountLimitsMemoryMaxStreamBytes limits
      , "storage_max_stream_bytes" .= accountLimitsStorageMaxStreamBytes limits
      , "max_bytes_required" .= accountLimitsMaxBytesRequired limits
      ]

accountTierPairs :: AccountTier -> [Pair]
accountTierPairs tier =
  [ "memory" .= accountTierMemory tier
  , "storage" .= accountTierStorage tier
  , "reserved_memory" .= accountTierReservedMemory tier
  , "reserved_storage" .= accountTierReservedStorage tier
  , "streams" .= accountTierStreams tier
  , "consumers" .= accountTierConsumers tier
  , "limits" .= accountTierLimits tier
  ]

parseAccountTiers :: Object -> Parser [(BS.ByteString, AccountTier)]
parseAccountTiers obj = do
  tiers <- obj .:? "tiers"
  case tiers of
    Nothing ->
      pure []
    Just (Object tierMap) ->
      traverse parseTier (KeyMap.toList tierMap)
    Just invalid ->
      typeMismatch "AccountInfo.tiers" invalid
  where
    parseTier (name, value) =
      (,) (E.encodeUtf8 (Key.toText name)) <$> parseJSON value

accountTiersValue :: [(BS.ByteString, AccountTier)] -> Value
accountTiersValue =
  Object . KeyMap.fromList . fmap tierPair
  where
    tierPair (name, tier) =
      (Key.fromText (E.decodeUtf8 name), toJSON tier)

parseOptionalByteStringField :: Object -> Key.Key -> Parser (Maybe BS.ByteString)
parseOptionalByteStringField obj key = do
  value <- obj .:? key
  traverse parseByteString value

maybeByteStringPair :: Key.Key -> Maybe BS.ByteString -> Maybe Pair
maybeByteStringPair key =
  fmap ((key .=) . byteStringToJSON)

maybePair :: ToJSON value => Key.Key -> Maybe value -> Maybe Pair
maybePair key =
  fmap (key .=)