{-# LANGUAGE OverloadedStrings #-} module JetStream.KeyValue.Types ( KeyValueAPI (..) , KeyValueBucket (..) , KeyValueBucketName , KeyValueKey , KeyValuePattern , KeyValueValue , KeyValueRevision , KeyValueOperation (..) , KeyValueEntry (..) , KeyValueConfig (..) , KeyValueConfigOption , KeyValueStatus (..) , KeyValueError (..) , KeyValueDeleteOption , KeyValueDeleteConfig (..) , KeyValueWatchOption , KeyValueWatchConfig (..) , KeyValueWatcher (..) , KeyValueWatchBatch (..) , KeyValuePurgeDeletesOption , KeyValuePurgeDeletesConfig (..) , keyValueConfig , validateKeyValueConfig , validateKeyValueBucketName , validateKeyValueKey , validateKeyValuePattern , keyValueStreamName , keyValueSubjectPrefix , keyValueSubject , keyValuePatternSubject , keyValueDuplicateWindow , keyValueStreamOptions , keyValueEntryFromStreamMessage , keyValueEntryFromMessage , keyValueDeleteConfig , keyValueWatchConfig , validateKeyValueWatchConfig , keyValuePurgeDeletesConfig , withKeyValueDescription , withKeyValueMaxValueSize , withKeyValueHistory , withKeyValueTTL , withKeyValueMaxBytes , withKeyValueStorage , withKeyValueReplicas , withKeyValueCompression , withKeyValueLastRevision , withKeyValueIncludeHistory , withKeyValueUpdatesOnly , withKeyValueIgnoreDeletes , withKeyValueMetadataOnly , withKeyValueDeleteMarkersOlderThan ) where import Control.Concurrent.STM (TVar) import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as BC import Data.Char ( isAsciiLower , isAsciiUpper , isDigit , toLower ) import Data.Int (Int32) import Data.Maybe (catMaybes, fromMaybe) import Data.Time.Clock (NominalDiffTime, UTCTime) import Data.Word (Word64) import JetStream.Error (JetStreamError) import JetStream.Message.Types ( FetchOption , Message , OrderedConsumer , PullStatus , messageHeaders , messageMetadata , messageMetadataNumPending , messageMetadataStreamSequence , messageMetadataTimestamp , messagePayload , messageSubject ) import JetStream.Stream.Types ( StorageType (FileStorage) , StreamCompression (S2Compression) , StreamConfigOption , StreamMessage , streamMessageHeadersRaw , streamMessagePayload , streamMessageSequence , streamMessageSubject , streamMessageTime , withAllowDirect , withAllowRollup , withCompression , withDenyDelete , withDescription , withDiscard , withDuplicateWindow , withMaxAge , withMaxBytes , withMaxConsumers , withMaxMessageSize , withMaxMessages , withMaxMessagesPerSubject , withReplicas , withRetention , withStorage ) import JetStream.Types ( CallOption , DiscardPolicy (DiscardNew) , JetStreamRequestOption , RetentionPolicy (LimitsPolicy) , applyCallOptions ) import Parser.Attoparsec (parseHeaderBlock) type KeyValueBucketName = BS.ByteString type KeyValueKey = BS.ByteString type KeyValuePattern = BS.ByteString type KeyValueValue = BS.ByteString type KeyValueRevision = Word64 data KeyValueAPI = KeyValueAPI { createKeyValueBucket :: KeyValueBucketName -> [KeyValueConfigOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueBucket) , updateKeyValueBucket :: KeyValueBucketName -> [KeyValueConfigOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueBucket) , createOrUpdateKeyValueBucket :: KeyValueBucketName -> [KeyValueConfigOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueBucket) , lookupKeyValueBucket :: KeyValueBucketName -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueBucket) , deleteKeyValueBucket :: KeyValueBucketName -> [JetStreamRequestOption] -> IO (Either KeyValueError ()) , listKeyValueBuckets :: [JetStreamRequestOption] -> IO (Either KeyValueError [KeyValueBucket]) , listKeyValueStatuses :: [JetStreamRequestOption] -> IO (Either KeyValueError [KeyValueStatus]) , getKeyValueStatus :: KeyValueBucket -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueStatus) , getKeyValueEntry :: KeyValueBucket -> KeyValueKey -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueEntry) , getKeyValueEntryRevision :: KeyValueBucket -> KeyValueKey -> KeyValueRevision -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueEntry) , putKeyValueEntry :: KeyValueBucket -> KeyValueKey -> KeyValueValue -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueRevision) , createKeyValueEntry :: KeyValueBucket -> KeyValueKey -> KeyValueValue -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueRevision) , updateKeyValueEntry :: KeyValueBucket -> KeyValueKey -> KeyValueValue -> KeyValueRevision -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueRevision) , deleteKeyValueEntry :: KeyValueBucket -> KeyValueKey -> [KeyValueDeleteOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueRevision) , purgeKeyValueEntry :: KeyValueBucket -> KeyValueKey -> [KeyValueDeleteOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueRevision) , watchKeyValues :: KeyValueBucket -> [KeyValuePattern] -> [KeyValueWatchOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueWatcher) , fetchKeyValueWatch :: KeyValueWatcher -> [FetchOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError KeyValueWatchBatch) , stopKeyValueWatch :: KeyValueWatcher -> [JetStreamRequestOption] -> IO (Either KeyValueError ()) , listKeyValueKeys :: KeyValueBucket -> [JetStreamRequestOption] -> IO (Either KeyValueError [KeyValueKey]) , getKeyValueHistory :: KeyValueBucket -> KeyValueKey -> [JetStreamRequestOption] -> IO (Either KeyValueError [KeyValueEntry]) , purgeDeletedKeyValueEntries :: KeyValueBucket -> [KeyValuePurgeDeletesOption] -> [JetStreamRequestOption] -> IO (Either KeyValueError ()) } newtype KeyValueBucket = KeyValueBucket { keyValueBucketName :: KeyValueBucketName } deriving (Eq, Ord, Show) data KeyValueOperation = KeyValuePut | KeyValueDelete | KeyValuePurge deriving (Eq, Show) data KeyValueEntry = KeyValueEntry { keyValueEntryBucket :: KeyValueBucketName , keyValueEntryKey :: KeyValueKey , keyValueEntryValue :: KeyValueValue , keyValueEntryRevision :: KeyValueRevision , keyValueEntryCreated :: UTCTime , keyValueEntryDelta :: Integer , keyValueEntryOperation :: KeyValueOperation } deriving (Eq, Show) data KeyValueConfig = KeyValueConfig { keyValueConfigBucket :: KeyValueBucketName , keyValueConfigDescription :: Maybe BS.ByteString , keyValueConfigMaxValueSize :: Int32 , keyValueConfigHistory :: Int , keyValueConfigTTL :: NominalDiffTime , keyValueConfigMaxBytes :: Integer , keyValueConfigStorage :: StorageType , keyValueConfigReplicas :: Int , keyValueConfigCompression :: Bool } deriving (Eq, Show) type KeyValueConfigOption = CallOption KeyValueConfig data KeyValueStatus = KeyValueStatus { keyValueStatusBucket :: KeyValueBucketName , keyValueStatusValues :: Integer , keyValueStatusBytes :: Integer , keyValueStatusConfig :: KeyValueConfig } deriving (Eq, Show) data KeyValueError = KeyValueJetStreamError JetStreamError | KeyValueInvalidBucketName KeyValueBucketName | KeyValueInvalidKey KeyValueKey | KeyValueInvalidPattern KeyValuePattern | KeyValueInvalidHistory Int | KeyValueInvalidMaxValueSize Int32 | KeyValueInvalidMaxBytes Integer | KeyValueInvalidTTL NominalDiffTime | KeyValueInvalidReplicas Int | KeyValueInvalidWatchOptions | KeyValueBucketNotFound KeyValueBucketName | KeyValueBucketExists KeyValueBucketName | KeyValueInvalidBucket KeyValueBucketName | KeyValueKeyNotFound KeyValueBucketName KeyValueKey | KeyValueKeyExists KeyValueBucketName KeyValueKey | KeyValueRevisionMismatch KeyValueBucketName KeyValueKey KeyValueRevision | KeyValueNoKeysFound KeyValueBucketName | KeyValueDecodeError String deriving (Eq, Show) newtype KeyValueDeleteConfig = KeyValueDeleteConfig { keyValueDeleteExpectedRevision :: Maybe KeyValueRevision } deriving (Eq, Show) type KeyValueDeleteOption = CallOption KeyValueDeleteConfig data KeyValueWatchConfig = KeyValueWatchConfig { keyValueWatchIncludeHistory :: Bool , keyValueWatchUpdatesOnly :: Bool , keyValueWatchIgnoreDeletes :: Bool , keyValueWatchMetadataOnly :: Bool } deriving (Eq, Show) type KeyValueWatchOption = CallOption KeyValueWatchConfig data KeyValueWatcher = KeyValueWatcher { keyValueWatcherBucket :: KeyValueBucket , keyValueWatcherConsumer :: OrderedConsumer , keyValueWatcherIgnoreDeletes :: Bool , keyValueWatcherInitialRemaining :: TVar Integer , keyValueWatcherInitialComplete :: TVar Bool } data KeyValueWatchBatch = KeyValueWatchBatch { keyValueWatchEntries :: [KeyValueEntry] , keyValueWatchInitialComplete :: Bool , keyValueWatchStatus :: Maybe PullStatus } deriving (Eq, Show) newtype KeyValuePurgeDeletesConfig = KeyValuePurgeDeletesConfig { keyValueDeleteMarkersOlderThan :: NominalDiffTime } deriving (Eq, Show) type KeyValuePurgeDeletesOption = CallOption KeyValuePurgeDeletesConfig keyValueConfig :: KeyValueBucketName -> [KeyValueConfigOption] -> KeyValueConfig keyValueConfig bucket options = normalizeKeyValueConfig $ applyCallOptions options KeyValueConfig { keyValueConfigBucket = bucket , keyValueConfigDescription = Nothing , keyValueConfigMaxValueSize = -1 , keyValueConfigHistory = 1 , keyValueConfigTTL = 0 , keyValueConfigMaxBytes = -1 , keyValueConfigStorage = FileStorage , keyValueConfigReplicas = 1 , keyValueConfigCompression = False } normalizeKeyValueConfig :: KeyValueConfig -> KeyValueConfig normalizeKeyValueConfig config = config { keyValueConfigMaxValueSize = defaultWhenZero (-1) (keyValueConfigMaxValueSize config) , keyValueConfigHistory = defaultWhenZero 1 (keyValueConfigHistory config) , keyValueConfigMaxBytes = defaultWhenZero (-1) (keyValueConfigMaxBytes config) , keyValueConfigReplicas = defaultWhenZero 1 (keyValueConfigReplicas config) } where defaultWhenZero fallback value | value == 0 = fallback | otherwise = value validateKeyValueConfig :: KeyValueConfig -> Either KeyValueError () validateKeyValueConfig config = do validateKeyValueBucketName (keyValueConfigBucket config) if keyValueConfigHistory config < 1 || keyValueConfigHistory config > 64 then Left (KeyValueInvalidHistory (keyValueConfigHistory config)) else Right () if keyValueConfigMaxValueSize config < (-1) then Left (KeyValueInvalidMaxValueSize (keyValueConfigMaxValueSize config)) else Right () if keyValueConfigMaxBytes config < (-1) then Left (KeyValueInvalidMaxBytes (keyValueConfigMaxBytes config)) else Right () if keyValueConfigTTL config < 0 then Left (KeyValueInvalidTTL (keyValueConfigTTL config)) else Right () if keyValueConfigReplicas config < 1 then Left (KeyValueInvalidReplicas (keyValueConfigReplicas config)) else Right () validateKeyValueBucketName :: KeyValueBucketName -> Either KeyValueError () validateKeyValueBucketName bucket | BS.null bucket = Left (KeyValueInvalidBucketName bucket) | BC.all isBucketCharacter bucket = Right () | otherwise = Left (KeyValueInvalidBucketName bucket) validateKeyValueKey :: KeyValueKey -> Either KeyValueError () validateKeyValueKey key | validKeyLike isKeyCharacter key = Right () | otherwise = Left (KeyValueInvalidKey key) validateKeyValuePattern :: KeyValuePattern -> Either KeyValueError () validateKeyValuePattern pattern' | validKeyLike isPatternCharacter pattern' && validGreaterThan pattern' = Right () | otherwise = Left (KeyValueInvalidPattern pattern') where validGreaterThan value = case BC.elemIndex '>' value of Nothing -> True Just index -> index == BS.length value - 1 validKeyLike :: (Char -> Bool) -> BS.ByteString -> Bool validKeyLike validCharacter value = not (BS.null value) && BC.head value /= '.' && BC.last value /= '.' && not (".." `BS.isInfixOf` value) && BC.all validCharacter value isBucketCharacter :: Char -> Bool isBucketCharacter char = isAsciiAlphaNumeric char || char == '_' || char == '-' isKeyCharacter :: Char -> Bool isKeyCharacter char = isAsciiAlphaNumeric char || char `elem` ("-/_=." :: String) isPatternCharacter :: Char -> Bool isPatternCharacter char = isKeyCharacter char || char == '*' || char == '>' isAsciiAlphaNumeric :: Char -> Bool isAsciiAlphaNumeric char = isAsciiLower char || isAsciiUpper char || isDigit char keyValueStreamName :: KeyValueBucketName -> BS.ByteString keyValueStreamName = BS.append "KV_" keyValueSubjectPrefix :: KeyValueBucketName -> BS.ByteString keyValueSubjectPrefix bucket = BS.concat ["$KV.", bucket, "."] keyValueSubject :: KeyValueBucketName -> KeyValueKey -> BS.ByteString keyValueSubject bucket = BS.append (keyValueSubjectPrefix bucket) keyValuePatternSubject :: KeyValueBucketName -> KeyValuePattern -> BS.ByteString keyValuePatternSubject = keyValueSubject keyValueStreamOptions :: KeyValueConfig -> [StreamConfigOption] keyValueStreamOptions config = catMaybes [ withDescription <$> keyValueConfigDescription config , Just (withRetention LimitsPolicy) , Just (withStorage (keyValueConfigStorage config)) , Just (withDiscard DiscardNew) , Just (withMaxConsumers (-1)) , Just (withMaxMessages (-1)) , Just (withMaxMessagesPerSubject (toInteger (keyValueConfigHistory config))) , Just (withMaxBytes (keyValueConfigMaxBytes config)) , Just (withMaxAge (keyValueConfigTTL config)) , Just (withMaxMessageSize (keyValueConfigMaxValueSize config)) , Just (withReplicas (keyValueConfigReplicas config)) , Just (withDuplicateWindow (keyValueDuplicateWindow config)) , Just (withDenyDelete True) , Just (withAllowRollup True) , Just (withAllowDirect True) , if keyValueConfigCompression config then Just (withCompression S2Compression) else Nothing ] keyValueDuplicateWindow :: KeyValueConfig -> NominalDiffTime keyValueDuplicateWindow config | keyValueConfigTTL config > 0 = min 120 (keyValueConfigTTL config) | otherwise = 120 keyValueEntryFromStreamMessage :: KeyValueBucket -> KeyValueKey -> StreamMessage -> Either KeyValueError KeyValueEntry keyValueEntryFromStreamMessage bucket key message | streamMessageSubject message /= expectedSubject = Left (KeyValueKeyNotFound bucketName key) | otherwise = do operation <- operationFromRawHeaders (streamMessageHeadersRaw message) Right KeyValueEntry { keyValueEntryBucket = bucketName , keyValueEntryKey = key , keyValueEntryValue = fromMaybe BS.empty (streamMessagePayload message) , keyValueEntryRevision = streamMessageSequence message , keyValueEntryCreated = streamMessageTime message , keyValueEntryDelta = 0 , keyValueEntryOperation = operation } where bucketName = keyValueBucketName bucket expectedSubject = keyValueSubject bucketName key keyValueEntryFromMessage :: KeyValueBucket -> Message -> Either KeyValueError KeyValueEntry keyValueEntryFromMessage bucket message = do metadata <- maybe (Left (KeyValueDecodeError "missing JetStream message metadata")) Right (messageMetadata message) key <- keyFromSubject bucket (messageSubject message) pure KeyValueEntry { keyValueEntryBucket = keyValueBucketName bucket , keyValueEntryKey = key , keyValueEntryValue = messagePayload message , keyValueEntryRevision = messageMetadataStreamSequence metadata , keyValueEntryCreated = messageMetadataTimestamp metadata , keyValueEntryDelta = messageMetadataNumPending metadata , keyValueEntryOperation = operationFromHeaders (messageHeaders message) } keyFromSubject :: KeyValueBucket -> BS.ByteString -> Either KeyValueError KeyValueKey keyFromSubject bucket subject = if prefix `BS.isPrefixOf` subject && BS.length subject > BS.length prefix then Right (BS.drop (BS.length prefix) subject) else Left (KeyValueDecodeError "key-value message subject does not match bucket") where prefix = keyValueSubjectPrefix (keyValueBucketName bucket) operationFromRawHeaders :: Maybe BS.ByteString -> Either KeyValueError KeyValueOperation operationFromRawHeaders Nothing = Right KeyValuePut operationFromRawHeaders (Just rawHeaders) = case parseHeaderBlock rawHeaders of Left err -> Left (KeyValueDecodeError err) Right headers -> Right (operationFromHeaders (Just headers)) operationFromHeaders :: Maybe [(BS.ByteString, BS.ByteString)] -> KeyValueOperation operationFromHeaders headers = case lookupHeader "KV-Operation" (fromMaybe [] headers) of Just "DEL" -> KeyValueDelete Just "PURGE" -> KeyValuePurge _ -> case lookupHeader "Nats-Marker-Reason" (fromMaybe [] headers) of Just "MaxAge" -> KeyValuePurge Just "Purge" -> KeyValuePurge Just "Remove" -> KeyValueDelete _ -> KeyValuePut lookupHeader :: BS.ByteString -> [(BS.ByteString, BS.ByteString)] -> Maybe BS.ByteString lookupHeader name = go where normalizedName = normalizeHeaderName name go [] = Nothing go ((headerName, value):headers) | normalizeHeaderName headerName == normalizedName = Just value | otherwise = go headers normalizeHeaderName :: BS.ByteString -> BS.ByteString normalizeHeaderName = BC.map toLower keyValueDeleteConfig :: [KeyValueDeleteOption] -> KeyValueDeleteConfig keyValueDeleteConfig options = applyCallOptions options (KeyValueDeleteConfig Nothing) keyValueWatchConfig :: [KeyValueWatchOption] -> KeyValueWatchConfig keyValueWatchConfig options = applyCallOptions options KeyValueWatchConfig { keyValueWatchIncludeHistory = False , keyValueWatchUpdatesOnly = False , keyValueWatchIgnoreDeletes = False , keyValueWatchMetadataOnly = False } validateKeyValueWatchConfig :: KeyValueWatchConfig -> Either KeyValueError () validateKeyValueWatchConfig config | keyValueWatchIncludeHistory config && keyValueWatchUpdatesOnly config = Left KeyValueInvalidWatchOptions | otherwise = Right () keyValuePurgeDeletesConfig :: [KeyValuePurgeDeletesOption] -> KeyValuePurgeDeletesConfig keyValuePurgeDeletesConfig options = applyCallOptions options (KeyValuePurgeDeletesConfig 1800) withKeyValueDescription :: BS.ByteString -> KeyValueConfigOption withKeyValueDescription description config = config { keyValueConfigDescription = Just description } withKeyValueMaxValueSize :: Int32 -> KeyValueConfigOption withKeyValueMaxValueSize maxValueSize config = config { keyValueConfigMaxValueSize = maxValueSize } withKeyValueHistory :: Int -> KeyValueConfigOption withKeyValueHistory history config = config { keyValueConfigHistory = history } withKeyValueTTL :: NominalDiffTime -> KeyValueConfigOption withKeyValueTTL ttl config = config { keyValueConfigTTL = ttl } withKeyValueMaxBytes :: Integer -> KeyValueConfigOption withKeyValueMaxBytes maxBytes config = config { keyValueConfigMaxBytes = maxBytes } withKeyValueStorage :: StorageType -> KeyValueConfigOption withKeyValueStorage storage config = config { keyValueConfigStorage = storage } withKeyValueReplicas :: Int -> KeyValueConfigOption withKeyValueReplicas replicas config = config { keyValueConfigReplicas = replicas } withKeyValueCompression :: Bool -> KeyValueConfigOption withKeyValueCompression compression config = config { keyValueConfigCompression = compression } withKeyValueLastRevision :: KeyValueRevision -> KeyValueDeleteOption withKeyValueLastRevision revision config = config { keyValueDeleteExpectedRevision = Just revision } withKeyValueIncludeHistory :: KeyValueWatchOption withKeyValueIncludeHistory config = config { keyValueWatchIncludeHistory = True } withKeyValueUpdatesOnly :: KeyValueWatchOption withKeyValueUpdatesOnly config = config { keyValueWatchUpdatesOnly = True } withKeyValueIgnoreDeletes :: KeyValueWatchOption withKeyValueIgnoreDeletes config = config { keyValueWatchIgnoreDeletes = True } withKeyValueMetadataOnly :: KeyValueWatchOption withKeyValueMetadataOnly config = config { keyValueWatchMetadataOnly = True } withKeyValueDeleteMarkersOlderThan :: NominalDiffTime -> KeyValuePurgeDeletesOption withKeyValueDeleteMarkersOlderThan olderThan config = config { keyValueDeleteMarkersOlderThan = olderThan }