diff --git a/client/JetStream/API.hs b/client/JetStream/API.hs --- a/client/JetStream/API.hs +++ b/client/JetStream/API.hs @@ -6,6 +6,7 @@ , publisher , messages , management + , keyValues , JetStreamRequestOption , withRequestTimeout , JetStreamApiError @@ -14,6 +15,7 @@ , apiErrorDescription , JetStreamError (..) , module JetStream.API.Consumer + , module JetStream.API.KeyValue , module JetStream.API.Management , module JetStream.API.Message , module JetStream.API.Publish @@ -21,6 +23,7 @@ ) where import JetStream.API.Consumer +import JetStream.API.KeyValue import JetStream.API.Management import JetStream.API.Message import JetStream.API.Publish @@ -35,6 +38,7 @@ import JetStream.Options ( JetStream , consumers + , keyValues , management , messages , publisher diff --git a/client/JetStream/API/KeyValue.hs b/client/JetStream/API/KeyValue.hs new file mode 100644 --- /dev/null +++ b/client/JetStream/API/KeyValue.hs @@ -0,0 +1,4 @@ +-- | JetStream key-value contract. +module JetStream.API.KeyValue (module JetStream.KeyValue.API) where + +import JetStream.KeyValue.API diff --git a/client/JetStream/Client.hs b/client/JetStream/Client.hs --- a/client/JetStream/Client.hs +++ b/client/JetStream/Client.hs @@ -13,6 +13,7 @@ import qualified API as Nats import JetStream.API (JetStream) import qualified JetStream.Consumer as Consumer +import qualified JetStream.KeyValue as KeyValue import qualified JetStream.Management as Management import qualified JetStream.Message as Message import JetStream.Options @@ -32,11 +33,15 @@ newJetStream :: Nats.Client -> [JetStreamOption] -> Either JetStreamConfigError JetStream newJetStream client options = do ctx <- tryNewJetStreamContext client options - let consumerAPI = Consumer.consumerAPI ctx + let streamAPI = Stream.streamAPI ctx + consumerAPI = Consumer.consumerAPI ctx + publishAPI = Publish.publishAPI ctx + messageAPI = Message.messageAPI ctx consumerAPI pure JetStream - { streams = Stream.streamAPI ctx + { streams = streamAPI , consumers = consumerAPI - , publisher = Publish.publishAPI ctx - , messages = Message.messageAPI ctx consumerAPI + , publisher = publishAPI + , messages = messageAPI , management = Management.managementAPI ctx + , keyValues = KeyValue.keyValueAPI streamAPI publishAPI messageAPI } diff --git a/internal/Plumbing/Parser/Attoparsec.hs b/internal/Plumbing/Parser/Attoparsec.hs --- a/internal/Plumbing/Parser/Attoparsec.hs +++ b/internal/Plumbing/Parser/Attoparsec.hs @@ -3,6 +3,7 @@ module Parser.Attoparsec ( parserApi , parserApiWithMessageLimit + , parseHeaderBlock ) where import Control.Applicative ((<|>)) @@ -33,6 +34,10 @@ parserApiWithMessageLimit :: Int -> ParserAPI ParsedMessage parserApiWithMessageLimit maximumMessageSize = ParserAPI (parseStep (max 1 maximumMessageSize)) + +parseHeaderBlock :: BS.ByteString -> Either String [(BS.ByteString, BS.ByteString)] +parseHeaderBlock = + A.parseOnly (headerBlockParser <* A.endOfInput) parseStep :: Int -> BS.ByteString -> ParseStep ParsedMessage parseStep maximumMessageSize bytes = diff --git a/jetstream/JetStream/KeyValue.hs b/jetstream/JetStream/KeyValue.hs new file mode 100644 --- /dev/null +++ b/jetstream/JetStream/KeyValue.hs @@ -0,0 +1,811 @@ +{-# LANGUAGE OverloadedStrings #-} + +module JetStream.KeyValue + ( keyValueAPI + , module JetStream.KeyValue.API + ) where + +import Control.Concurrent.STM + ( atomically + , modifyTVar' + , newTVarIO + , readTVar + , readTVarIO + , writeTVar + ) +import Control.Exception (finally) +import Control.Monad (foldM, unless, void, when) +import qualified Data.ByteString as BS +import Data.List (foldl', sort) +import Data.Maybe (catMaybes, fromMaybe, mapMaybe) +import Data.Time.Clock (diffUTCTime, getCurrentTime) +import JetStream.Consumer.Types + ( ConsumerFilter (ConsumerFilterSubject, ConsumerFilterSubjects) + , consumerInfoNumPending + ) +import JetStream.Error + ( JetStreamApiError (apiErrorCodeDetail) + , JetStreamError (JetStreamApiFailure) + ) +import JetStream.KeyValue.API +import JetStream.KeyValue.Types +import qualified JetStream.Message.API as Message +import JetStream.Message.Types + ( FetchWait (FetchNoWaitMicros) + , pullResponseMessages + , pullResponseStatus + , withFetchBatch + , withFetchWait + , withOrderedConsumerDeliverPolicy + , withOrderedConsumerFilter + , withOrderedConsumerHeadersOnly + ) +import qualified JetStream.Publish.API as Publish +import qualified JetStream.Stream.API as Stream +import JetStream.Stream.Types + ( StreamCompression (S2Compression) + , StreamConfig + , StreamInfo + , StreamMessageSelector (LastStreamMessageForSubject, StreamMessageBySequence) + , streamConfigAllowDirect + , streamConfigAllowRollup + , streamConfigCompression + , streamConfigDenyDelete + , streamConfigDescription + , streamConfigDiscard + , streamConfigDuplicateWindow + , streamConfigMaxAge + , streamConfigMaxBytes + , streamConfigMaxConsumers + , streamConfigMaxMessageSize + , streamConfigMaxMessages + , streamConfigMaxMessagesPerSubject + , streamConfigName + , streamConfigReplicas + , streamConfigRetention + , streamConfigStorage + , streamConfigSubjects + , streamInfoConfig + , streamInfoState + , streamListOffset + , streamListStreams + , streamListTotal + , streamNamesOffset + , streamNamesStreams + , streamNamesTotal + , streamStateBytes + , streamStateMessages + ) +import JetStream.Types + ( DeliverPolicy (DeliverAll, DeliverLastPerSubject, DeliverNew) + , DiscardPolicy (DiscardNew) + , JetStreamRequestOption + , RetentionPolicy (LimitsPolicy) + ) + +keyValueAPI :: Stream.StreamAPI -> Publish.PublishAPI -> Message.MessageAPI -> KeyValueAPI +keyValueAPI streamAPI publishAPI messageAPI = + KeyValueAPI + { createKeyValueBucket = createBucket streamAPI + , updateKeyValueBucket = configureBucket (Stream.update streamAPI) + , createOrUpdateKeyValueBucket = configureBucket (Stream.createOrUpdate streamAPI) + , lookupKeyValueBucket = lookupBucket streamAPI + , deleteKeyValueBucket = deleteBucket streamAPI + , listKeyValueBuckets = listBuckets streamAPI + , listKeyValueStatuses = listStatuses streamAPI + , getKeyValueStatus = bucketStatus streamAPI + , getKeyValueEntry = getLatestEntry streamAPI + , getKeyValueEntryRevision = getEntryRevision streamAPI + , putKeyValueEntry = putEntry publishAPI + , createKeyValueEntry = createEntry streamAPI publishAPI + , updateKeyValueEntry = updateEntry publishAPI + , deleteKeyValueEntry = deleteEntry publishAPI + , purgeKeyValueEntry = purgeEntry publishAPI + , watchKeyValues = createWatcher messageAPI + , fetchKeyValueWatch = fetchWatcher + , stopKeyValueWatch = stopWatcher + , listKeyValueKeys = listKeys messageAPI + , getKeyValueHistory = entryHistory messageAPI + , purgeDeletedKeyValueEntries = purgeDeletedEntries streamAPI messageAPI + } + +createBucket + :: Stream.StreamAPI + -> KeyValueBucketName + -> [KeyValueConfigOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueBucket) +createBucket streamAPI bucket options requestOptions = + case validateKeyValueConfig config of + Left err -> pure (Left err) + Right () -> do + created <- Stream.create streamAPI streamName [subject] + (keyValueStreamOptions config) requestOptions + case created of + Right info -> + pure (validateBucketInfo handle info >> Right handle) + Left err + | isApiError 10058 err -> existingBucket + | otherwise -> pure (Left (mapBucketError bucket err)) + where + config = keyValueConfig bucket options + handle = KeyValueBucket bucket + streamName = keyValueStreamName bucket + subject = keyValuePatternSubject bucket ">" + existingBucket = do + existing <- Stream.info streamAPI streamName requestOptions + pure $ do + info <- either (Left . mapBucketError bucket) Right existing + validateBucketInfo handle info + if compatibleBucketConfig config (streamInfoConfig info) + then Right handle + else Left (KeyValueBucketExists bucket) + +configureBucket + :: ( BS.ByteString + -> [BS.ByteString] + -> [Stream.StreamConfigOption] + -> [JetStreamRequestOption] + -> IO (Either JetStreamError StreamInfo) + ) + -> KeyValueBucketName + -> [KeyValueConfigOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueBucket) +configureBucket configure bucket options requestOptions = + case validateKeyValueConfig config of + Left err -> pure (Left err) + Right () -> do + result <- configure streamName [subject] + (keyValueStreamOptions config) requestOptions + pure $ do + info <- either (Left . mapBucketError bucket) Right result + validateBucketInfo handle info + Right handle + where + config = keyValueConfig bucket options + handle = KeyValueBucket bucket + streamName = keyValueStreamName bucket + subject = keyValuePatternSubject bucket ">" + +lookupBucket + :: Stream.StreamAPI + -> KeyValueBucketName + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueBucket) +lookupBucket streamAPI bucket requestOptions = + case validateKeyValueBucketName bucket of + Left err -> pure (Left err) + Right () -> do + info <- Stream.info streamAPI (keyValueStreamName bucket) requestOptions + pure $ do + detail <- either (Left . mapBucketError bucket) Right info + let handle = KeyValueBucket bucket + validateBucketInfo handle detail + Right handle + +deleteBucket + :: Stream.StreamAPI + -> KeyValueBucketName + -> [JetStreamRequestOption] + -> IO (Either KeyValueError ()) +deleteBucket streamAPI bucket requestOptions = + case validateKeyValueBucketName bucket of + Left err -> pure (Left err) + Right () -> do + deleted <- Stream.delete streamAPI (keyValueStreamName bucket) requestOptions + pure $ do + response <- either (Left . mapBucketError bucket) Right deleted + if Stream.deleteStreamSuccess response + then Right () + else Left (KeyValueDecodeError "JetStream did not delete key-value bucket") + +listBuckets + :: Stream.StreamAPI + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [KeyValueBucket]) +listBuckets streamAPI requestOptions = + fmap (fmap (map KeyValueBucket . mapMaybe bucketFromStreamName)) + (listAllKeyValueStreamNames streamAPI requestOptions) + +listStatuses + :: Stream.StreamAPI + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [KeyValueStatus]) +listStatuses streamAPI requestOptions = + go 0 [] + where + go offset reversedStatuses = do + response <- Stream.list streamAPI + [ Stream.withStreamListOffset offset + , Stream.withStreamListSubject "$KV.*.>" + ] + requestOptions + case response of + Left err -> pure (Left (KeyValueJetStreamError err)) + Right page -> + case catMaybes <$> traverse listedStatus (streamListStreams page) of + Left err -> pure (Left err) + Right pageStatuses -> do + let accumulated = reverse pageStatuses ++ reversedStatuses + nextOffset = streamListOffset page + length (streamListStreams page) + if null (streamListStreams page) || nextOffset >= streamListTotal page + then pure (Right (reverse accumulated)) + else go nextOffset accumulated + + listedStatus info = + case bucketFromStreamName (streamConfigName (streamInfoConfig info)) of + Nothing -> Right Nothing + Just bucketName -> do + let bucket = KeyValueBucket bucketName + validateBucketInfo bucket info + Just <$> statusFromInfo bucket info + +bucketStatus + :: Stream.StreamAPI + -> KeyValueBucket + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueStatus) +bucketStatus streamAPI bucket requestOptions = do + info <- Stream.info streamAPI + (keyValueStreamName (keyValueBucketName bucket)) requestOptions + pure $ do + detail <- either (Left . mapBucketError (keyValueBucketName bucket)) Right info + validateBucketInfo bucket detail + statusFromInfo bucket detail + +getLatestEntry + :: Stream.StreamAPI + -> KeyValueBucket + -> KeyValueKey + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueEntry) +getLatestEntry streamAPI bucket key requestOptions = do + result <- getRawEntry streamAPI bucket key + (LastStreamMessageForSubject (keyValueSubject bucketName key)) requestOptions + pure (result >>= requireValueEntry bucket key) + where + bucketName = keyValueBucketName bucket + +getEntryRevision + :: Stream.StreamAPI + -> KeyValueBucket + -> KeyValueKey + -> KeyValueRevision + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueEntry) +getEntryRevision streamAPI bucket key revision requestOptions = do + result <- getRawEntry streamAPI bucket key selector requestOptions + pure (result >>= requireValueEntry bucket key) + where + selector + | revision == 0 = LastStreamMessageForSubject + (keyValueSubject (keyValueBucketName bucket) key) + | otherwise = StreamMessageBySequence revision + +getRawEntry + :: Stream.StreamAPI + -> KeyValueBucket + -> KeyValueKey + -> StreamMessageSelector + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueEntry) +getRawEntry streamAPI bucket key selector requestOptions = + case validateKeyValueKey key of + Left err -> pure (Left err) + Right () -> do + message <- Stream.getMessage streamAPI + (keyValueStreamName bucketName) selector requestOptions + pure $ do + stored <- either (Left . mapEntryReadError bucketName key) Right message + keyValueEntryFromStreamMessage bucket key stored + where + bucketName = keyValueBucketName bucket + +requireValueEntry + :: KeyValueBucket + -> KeyValueKey + -> KeyValueEntry + -> Either KeyValueError KeyValueEntry +requireValueEntry bucket key entry = + case keyValueEntryOperation entry of + KeyValuePut -> Right entry + _ -> Left (KeyValueKeyNotFound (keyValueBucketName bucket) key) + +putEntry + :: Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> KeyValueValue + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +putEntry publishAPI bucket key value = + publishEntry publishAPI bucket key value Nothing [] + +createEntry + :: Stream.StreamAPI + -> Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> KeyValueValue + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +createEntry streamAPI publishAPI bucket key value requestOptions = do + created <- updateEntry publishAPI bucket key value 0 requestOptions + case created of + Right revision -> pure (Right revision) + Left (KeyValueRevisionMismatch {}) -> do + existing <- getRawEntry streamAPI bucket key + (LastStreamMessageForSubject (keyValueSubject bucketName key)) requestOptions + case existing of + Right entry + | keyValueEntryOperation entry /= KeyValuePut -> + updateEntry publishAPI bucket key value + (keyValueEntryRevision entry) requestOptions + | otherwise -> + pure (Left (KeyValueKeyExists bucketName key)) + Left (KeyValueKeyNotFound _ _) -> + updateEntry publishAPI bucket key value 0 requestOptions + >>= \retry -> pure $ + case retry of + Left (KeyValueRevisionMismatch {}) -> + Left (KeyValueKeyExists bucketName key) + other -> other + Left err -> + pure (Left err) + Left err -> pure (Left err) + where + bucketName = keyValueBucketName bucket + +updateEntry + :: Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> KeyValueValue + -> KeyValueRevision + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +updateEntry publishAPI bucket key value revision = + publishEntry publishAPI bucket key value (Just revision) + [Publish.withPublishExpectation (Publish.ExpectedLastSubjectSequence revision)] + +publishEntry + :: Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> KeyValueValue + -> Maybe KeyValueRevision + -> [Publish.PublishOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +publishEntry publishAPI bucket key value expectedRevision options requestOptions = + case validateKeyValueKey key of + Left err -> pure (Left err) + Right () -> do + published <- Publish.publish publishAPI + (keyValueSubject bucketName key) + value + options + requestOptions + pure $ either + (Left . mapEntryWriteError bucketName key (fromMaybe 0 expectedRevision)) + (Right . Publish.publishAckSequence) + published + where + bucketName = keyValueBucketName bucket + +deleteEntry + :: Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> [KeyValueDeleteOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +deleteEntry publishAPI = + publishDeleteMarker publishAPI KeyValueDelete + +purgeEntry + :: Publish.PublishAPI + -> KeyValueBucket + -> KeyValueKey + -> [KeyValueDeleteOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +purgeEntry publishAPI = + publishDeleteMarker publishAPI KeyValuePurge + +publishDeleteMarker + :: Publish.PublishAPI + -> KeyValueOperation + -> KeyValueBucket + -> KeyValueKey + -> [KeyValueDeleteOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueRevision) +publishDeleteMarker publishAPI operation bucket key options = + publishEntry publishAPI bucket key BS.empty + (keyValueDeleteExpectedRevision config) publishOptions + where + config = keyValueDeleteConfig options + operationHeaders = + case operation of + KeyValueDelete -> [("KV-Operation", "DEL")] + KeyValuePurge -> + [ ("KV-Operation", "PURGE") + , ("Nats-Rollup", "sub") + ] + KeyValuePut -> [] + publishOptions = + Publish.withHeaders operationHeaders + : maybe [] + (\revision -> + [Publish.withPublishExpectation + (Publish.ExpectedLastSubjectSequence revision)]) + (keyValueDeleteExpectedRevision config) + +createWatcher + :: Message.MessageAPI + -> KeyValueBucket + -> [KeyValuePattern] + -> [KeyValueWatchOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueWatcher) +createWatcher messageAPI bucket patterns options requestOptions = + case validateWatcherInputs actualPatterns config of + Left err -> pure (Left err) + Right () -> do + created <- Message.createOrderedConsumer messageAPI streamName + consumerOptions requestOptions + case created of + Left err -> pure (Left (mapBucketError bucketName err)) + Right consumer -> do + info <- Message.orderedConsumerInfo consumer requestOptions + case info of + Left err -> do + void (Message.stopOrderedConsumer consumer requestOptions) + pure (Left (mapBucketError bucketName err)) + Right detail -> do + let initialPending + | keyValueWatchUpdatesOnly config = 0 + | otherwise = consumerInfoNumPending detail + remaining <- newTVarIO initialPending + complete <- newTVarIO (initialPending == 0) + pure (Right KeyValueWatcher + { keyValueWatcherBucket = bucket + , keyValueWatcherConsumer = consumer + , keyValueWatcherIgnoreDeletes = keyValueWatchIgnoreDeletes config + , keyValueWatcherInitialRemaining = remaining + , keyValueWatcherInitialComplete = complete + }) + where + bucketName = keyValueBucketName bucket + streamName = keyValueStreamName bucketName + actualPatterns + | null patterns = [">"] + | otherwise = patterns + config = keyValueWatchConfig options + filters = map (keyValuePatternSubject bucketName) actualPatterns + consumerFilter = + case filters of + [subject] -> ConsumerFilterSubject subject + subjects -> ConsumerFilterSubjects subjects + deliverPolicy + | keyValueWatchUpdatesOnly config = DeliverNew + | keyValueWatchIncludeHistory config = DeliverAll + | otherwise = DeliverLastPerSubject + consumerOptions = + [ withOrderedConsumerDeliverPolicy deliverPolicy + , withOrderedConsumerFilter consumerFilter + ] ++ + [ withOrderedConsumerHeadersOnly True + | keyValueWatchMetadataOnly config + ] + +validateWatcherInputs + :: [KeyValuePattern] + -> KeyValueWatchConfig + -> Either KeyValueError () +validateWatcherInputs patterns config = do + mapM_ validateKeyValuePattern patterns + validateKeyValueWatchConfig config + +fetchWatcher + :: KeyValueWatcher + -> [Message.FetchOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError KeyValueWatchBatch) +fetchWatcher watcher options requestOptions = do + response <- Message.fetchOrdered + (keyValueWatcherConsumer watcher) options requestOptions + case response of + Left err -> pure (Left (KeyValueJetStreamError err)) + Right result -> do + let rawMessages = pullResponseMessages result + case traverse + (keyValueEntryFromMessage (keyValueWatcherBucket watcher)) + rawMessages of + Left err -> pure (Left err) + Right entries -> do + updateInitialProgress watcher entries + complete <- readTVarIO (keyValueWatcherInitialComplete watcher) + pure (Right KeyValueWatchBatch + { keyValueWatchEntries = filterWatcherEntries watcher entries + , keyValueWatchInitialComplete = complete + , keyValueWatchStatus = pullResponseStatus result + }) + +updateInitialProgress :: KeyValueWatcher -> [KeyValueEntry] -> IO () +updateInitialProgress watcher entries = + atomically $ do + complete <- readTVar (keyValueWatcherInitialComplete watcher) + unless complete $ do + modifyTVar' (keyValueWatcherInitialRemaining watcher) + (max 0 . subtract received) + remaining <- readTVar (keyValueWatcherInitialRemaining watcher) + when (remaining == 0 || finalPending == Just 0) $ + writeTVar (keyValueWatcherInitialComplete watcher) True + where + received = toInteger (length entries) + finalPending = foldl' + (\_ entry -> Just (keyValueEntryDelta entry)) Nothing entries + +filterWatcherEntries :: KeyValueWatcher -> [KeyValueEntry] -> [KeyValueEntry] +filterWatcherEntries watcher + | keyValueWatcherIgnoreDeletes watcher = + filter ((== KeyValuePut) . keyValueEntryOperation) + | otherwise = id + +stopWatcher + :: KeyValueWatcher + -> [JetStreamRequestOption] + -> IO (Either KeyValueError ()) +stopWatcher watcher requestOptions = + fmap (either (Left . KeyValueJetStreamError) Right) + (Message.stopOrderedConsumer + (keyValueWatcherConsumer watcher) requestOptions) + +listKeys + :: Message.MessageAPI + -> KeyValueBucket + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [KeyValueKey]) +listKeys messageAPI bucket requestOptions = + withInitialEntries messageAPI bucket [] + [withKeyValueIgnoreDeletes, withKeyValueMetadataOnly] + requestOptions $ \entries -> + let keys = compact (sort (map keyValueEntryKey entries)) + in if null keys + then Left (KeyValueNoKeysFound (keyValueBucketName bucket)) + else Right keys + +entryHistory + :: Message.MessageAPI + -> KeyValueBucket + -> KeyValueKey + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [KeyValueEntry]) +entryHistory messageAPI bucket key requestOptions = + case validateKeyValueKey key of + Left err -> pure (Left err) + Right () -> + withInitialEntries messageAPI bucket [key] + [withKeyValueIncludeHistory] requestOptions $ \entries -> + if null entries + then Left (KeyValueKeyNotFound (keyValueBucketName bucket) key) + else Right entries + +withInitialEntries + :: Message.MessageAPI + -> KeyValueBucket + -> [KeyValuePattern] + -> [KeyValueWatchOption] + -> [JetStreamRequestOption] + -> ([KeyValueEntry] -> Either KeyValueError value) + -> IO (Either KeyValueError value) +withInitialEntries messageAPI bucket patterns options requestOptions use = do + watcherResult <- createWatcher messageAPI bucket patterns options requestOptions + case watcherResult of + Left err -> pure (Left err) + Right watcher -> + fmap (>>= use) + (collectInitialEntries watcher requestOptions) + `finally` void (stopWatcher watcher requestOptions) + +collectInitialEntries + :: KeyValueWatcher + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [KeyValueEntry]) +collectInitialEntries watcher requestOptions = + go [] + where + go reversedEntries = do + batch <- fetchWatcher watcher + [withFetchBatch 256, withFetchWait (FetchNoWaitMicros 100000)] + requestOptions + case batch of + Left err -> pure (Left err) + Right result -> do + let accumulated = + reverse (keyValueWatchEntries result) ++ reversedEntries + if keyValueWatchInitialComplete result + then pure (Right (reverse accumulated)) + else case keyValueWatchStatus result of + Nothing -> go accumulated + Just _ -> pure (Left + (KeyValueDecodeError "key-value watcher ended before initial values arrived")) + +purgeDeletedEntries + :: Stream.StreamAPI + -> Message.MessageAPI + -> KeyValueBucket + -> [KeyValuePurgeDeletesOption] + -> [JetStreamRequestOption] + -> IO (Either KeyValueError ()) +purgeDeletedEntries streamAPI messageAPI bucket options requestOptions = do + entriesResult <- withInitialEntries messageAPI bucket [] + [withKeyValueMetadataOnly] requestOptions Right + case entriesResult of + Left err -> pure (Left err) + Right entries -> do + now <- getCurrentTime + foldM (purgeMarker now) (Right ()) (filter isDeleteMarker entries) + where + configuredThreshold = + keyValueDeleteMarkersOlderThan (keyValuePurgeDeletesConfig options) + threshold + | configuredThreshold == 0 = 1800 + | otherwise = configuredThreshold + isDeleteMarker entry = + keyValueEntryOperation entry /= KeyValuePut + purgeMarker _ (Left err) _ = pure (Left err) + purgeMarker now (Right ()) entry = do + let age = diffUTCTime now (keyValueEntryCreated entry) + keep + | threshold > 0 && age < threshold = 1 + | otherwise = 0 + purged <- Stream.purge streamAPI + (keyValueStreamName (keyValueBucketName bucket)) + [ Stream.withPurgeSubject + (keyValueSubject (keyValueBucketName bucket) (keyValueEntryKey entry)) + , Stream.withPurgeKeep keep + ] + requestOptions + pure $ do + response <- either (Left . KeyValueJetStreamError) Right purged + if Stream.purgeStreamSuccess response + then Right () + else Left (KeyValueDecodeError "JetStream did not purge key-value tombstone") + +listAllKeyValueStreamNames + :: Stream.StreamAPI + -> [JetStreamRequestOption] + -> IO (Either KeyValueError [BS.ByteString]) +listAllKeyValueStreamNames streamAPI requestOptions = + go 0 [] + where + go offset reversedNames = do + response <- Stream.names streamAPI + [ Stream.withStreamListOffset offset + , Stream.withStreamListSubject "$KV.*.>" + ] + requestOptions + case response of + Left err -> pure (Left (KeyValueJetStreamError err)) + Right page -> do + let pageNames = streamNamesStreams page + accumulated = reverse pageNames ++ reversedNames + nextOffset = streamNamesOffset page + length pageNames + if null pageNames || nextOffset >= streamNamesTotal page + then pure (Right (reverse accumulated)) + else go nextOffset accumulated + +bucketFromStreamName :: BS.ByteString -> Maybe KeyValueBucketName +bucketFromStreamName streamName = + if "KV_" `BS.isPrefixOf` streamName + then let bucket = BS.drop 3 streamName + in case validateKeyValueBucketName bucket of + Right () -> Just bucket + Left _ -> Nothing + else Nothing + +validateBucketInfo + :: KeyValueBucket + -> StreamInfo + -> Either KeyValueError () +validateBucketInfo bucket info + | streamConfigMaxMessagesPerSubject (streamInfoConfig info) < 1 = + Left (KeyValueInvalidBucket (keyValueBucketName bucket)) + | otherwise = Right () + +statusFromInfo + :: KeyValueBucket + -> StreamInfo + -> Either KeyValueError KeyValueStatus +statusFromInfo bucket info + | history > toInteger (maxBound :: Int) = + Left (KeyValueInvalidBucket bucketName) + | otherwise = + Right KeyValueStatus + { keyValueStatusBucket = bucketName + , keyValueStatusValues = streamStateMessages state + , keyValueStatusBytes = streamStateBytes state + , keyValueStatusConfig = KeyValueConfig + { keyValueConfigBucket = bucketName + , keyValueConfigDescription = streamConfigDescription config + , keyValueConfigMaxValueSize = streamConfigMaxMessageSize config + , keyValueConfigHistory = fromInteger history + , keyValueConfigTTL = streamConfigMaxAge config + , keyValueConfigMaxBytes = streamConfigMaxBytes config + , keyValueConfigStorage = streamConfigStorage config + , keyValueConfigReplicas = streamConfigReplicas config + , keyValueConfigCompression = streamConfigCompression config == S2Compression + } + } + where + bucketName = keyValueBucketName bucket + config = streamInfoConfig info + state = streamInfoState info + history = streamConfigMaxMessagesPerSubject config + +compatibleBucketConfig :: KeyValueConfig -> StreamConfig -> Bool +compatibleBucketConfig expected actual = + normalizeDescription (keyValueConfigDescription expected) + == normalizeDescription (streamConfigDescription actual) + && streamConfigSubjects actual + == Just [keyValuePatternSubject (keyValueConfigBucket expected) ">"] + && streamConfigRetention actual == LimitsPolicy + && streamConfigDiscard actual == DiscardNew + && streamConfigMaxConsumers actual == (-1) + && streamConfigMaxMessages actual == (-1) + && streamConfigMaxMessagesPerSubject actual + == toInteger (keyValueConfigHistory expected) + && streamConfigMaxBytes actual == keyValueConfigMaxBytes expected + && streamConfigMaxAge actual == keyValueConfigTTL expected + && streamConfigMaxMessageSize actual == keyValueConfigMaxValueSize expected + && streamConfigStorage actual == keyValueConfigStorage expected + && streamConfigReplicas actual == keyValueConfigReplicas expected + && streamConfigDuplicateWindow actual + == Just (keyValueDuplicateWindow expected) + && streamConfigDenyDelete actual + && streamConfigAllowRollup actual + && streamConfigAllowDirect actual + && (streamConfigCompression actual == S2Compression) + == keyValueConfigCompression expected + where + normalizeDescription = fromMaybe BS.empty + +mapBucketError :: KeyValueBucketName -> JetStreamError -> KeyValueError +mapBucketError bucket err + | isApiError 10059 err = KeyValueBucketNotFound bucket + | otherwise = KeyValueJetStreamError err + +mapEntryReadError + :: KeyValueBucketName + -> KeyValueKey + -> JetStreamError + -> KeyValueError +mapEntryReadError bucket key err + | isApiError 10037 err = KeyValueKeyNotFound bucket key + | isApiError 10059 err = KeyValueBucketNotFound bucket + | otherwise = KeyValueJetStreamError err + +mapEntryWriteError + :: KeyValueBucketName + -> KeyValueKey + -> KeyValueRevision + -> JetStreamError + -> KeyValueError +mapEntryWriteError bucket key revision err + | isApiError 10071 err = KeyValueRevisionMismatch bucket key revision + | isApiError 10059 err = KeyValueBucketNotFound bucket + | otherwise = KeyValueJetStreamError err + +isApiError :: Int -> JetStreamError -> Bool +isApiError code (JetStreamApiFailure err) = + apiErrorCodeDetail err == code +isApiError _ _ = False + +compact :: Eq value => [value] -> [value] +compact [] = [] +compact (value:values) = + value : compact (dropWhile (== value) values) diff --git a/jetstream/JetStream/KeyValue/API.hs b/jetstream/JetStream/KeyValue/API.hs new file mode 100644 --- /dev/null +++ b/jetstream/JetStream/KeyValue/API.hs @@ -0,0 +1,81 @@ +module JetStream.KeyValue.API + ( KeyValueAPI + , createKeyValueBucket + , updateKeyValueBucket + , createOrUpdateKeyValueBucket + , lookupKeyValueBucket + , deleteKeyValueBucket + , listKeyValueBuckets + , listKeyValueStatuses + , getKeyValueStatus + , getKeyValueEntry + , getKeyValueEntryRevision + , putKeyValueEntry + , createKeyValueEntry + , updateKeyValueEntry + , deleteKeyValueEntry + , purgeKeyValueEntry + , watchKeyValues + , fetchKeyValueWatch + , stopKeyValueWatch + , listKeyValueKeys + , getKeyValueHistory + , purgeDeletedKeyValueEntries + , KeyValueBucket + , KeyValueBucketName + , keyValueBucketName + , KeyValueKey + , KeyValuePattern + , KeyValueValue + , KeyValueRevision + , KeyValueOperation (..) + , KeyValueEntry + , keyValueEntryBucket + , keyValueEntryKey + , keyValueEntryValue + , keyValueEntryRevision + , keyValueEntryCreated + , keyValueEntryDelta + , keyValueEntryOperation + , KeyValueConfig + , keyValueConfigBucket + , keyValueConfigDescription + , keyValueConfigMaxValueSize + , keyValueConfigHistory + , keyValueConfigTTL + , keyValueConfigMaxBytes + , keyValueConfigStorage + , keyValueConfigReplicas + , keyValueConfigCompression + , KeyValueConfigOption + , withKeyValueDescription + , withKeyValueMaxValueSize + , withKeyValueHistory + , withKeyValueTTL + , withKeyValueMaxBytes + , withKeyValueStorage + , withKeyValueReplicas + , withKeyValueCompression + , KeyValueStatus + , keyValueStatusBucket + , keyValueStatusValues + , keyValueStatusBytes + , keyValueStatusConfig + , KeyValueError (..) + , KeyValueDeleteOption + , withKeyValueLastRevision + , KeyValueWatchOption + , withKeyValueIncludeHistory + , withKeyValueUpdatesOnly + , withKeyValueIgnoreDeletes + , withKeyValueMetadataOnly + , KeyValueWatcher + , KeyValueWatchBatch + , keyValueWatchEntries + , keyValueWatchInitialComplete + , keyValueWatchStatus + , KeyValuePurgeDeletesOption + , withKeyValueDeleteMarkersOlderThan + ) where + +import JetStream.KeyValue.Types diff --git a/jetstream/JetStream/KeyValue/Types.hs b/jetstream/JetStream/KeyValue/Types.hs new file mode 100644 --- /dev/null +++ b/jetstream/JetStream/KeyValue/Types.hs @@ -0,0 +1,544 @@ +{-# 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 } diff --git a/jetstream/JetStream/Message.hs b/jetstream/JetStream/Message.hs --- a/jetstream/JetStream/Message.hs +++ b/jetstream/JetStream/Message.hs @@ -153,8 +153,8 @@ fetchOrderedMessages :: OrderedState -> [FetchOption] -> [JetStreamRequestOption] -> IO (Either JetStreamError PullResponse) fetchOrderedMessages state options requestOptions = do - resetResult <- resetOrderedConsumer state requestOptions - case resetResult of + consumerResult <- currentOrResetOrderedConsumer state requestOptions + case consumerResult of Left err -> pure (Left err) Right info -> do @@ -177,6 +177,27 @@ atomically $ writeTVar (orderedStateNextSequence state) (Just nextSequence) pure (Right response) + +currentOrResetOrderedConsumer + :: OrderedState + -> [JetStreamRequestOption] + -> IO (Either JetStreamError ConsumerInfo) +currentOrResetOrderedConsumer state requestOptions = do + stopped <- readTVarIO (orderedStateStopped state) + if stopped + then pure (Left JetStreamNoReply) + else do + currentName <- readTVarIO (orderedStateCurrentName state) + nextSequence <- readTVarIO (orderedStateNextSequence state) + case (currentName, nextSequence) of + (Just consumerName, Nothing) -> + consumerInfo + (orderedStateConsumers state) + (orderedStateStream state) + consumerName + requestOptions + _ -> + resetOrderedConsumer state requestOptions resetOrderedConsumer :: OrderedState -> [JetStreamRequestOption] -> IO (Either JetStreamError ConsumerInfo) resetOrderedConsumer state requestOptions = do diff --git a/jetstream/JetStream/Options.hs b/jetstream/JetStream/Options.hs --- a/jetstream/JetStream/Options.hs +++ b/jetstream/JetStream/Options.hs @@ -19,6 +19,7 @@ import qualified Client.API as Nats import qualified Data.ByteString as BS import JetStream.Consumer.API (ConsumerAPI) +import JetStream.KeyValue.API (KeyValueAPI) import JetStream.Management.API (ManagementAPI) import JetStream.Message.API (MessageAPI) import JetStream.Publish.API (PublishAPI) @@ -37,6 +38,7 @@ , publisher :: PublishAPI , messages :: MessageAPI , management :: ManagementAPI + , keyValues :: KeyValueAPI } data JetStreamConfig = JetStreamConfig diff --git a/jetstream/JetStream/Stream/API.hs b/jetstream/JetStream/Stream/API.hs --- a/jetstream/JetStream/Stream/API.hs +++ b/jetstream/JetStream/Stream/API.hs @@ -18,30 +18,43 @@ , RetentionPolicy (..) , StorageType (..) , DiscardPolicy (..) + , StreamCompression (..) , StreamConfig , streamConfigName , streamConfigSubjects + , streamConfigDescription , streamConfigRetention , streamConfigStorage , streamConfigDiscard + , streamConfigMaxConsumers , streamConfigMaxMessages + , streamConfigMaxMessagesPerSubject , streamConfigMaxBytes , streamConfigMaxAge , streamConfigMaxMessageSize , streamConfigReplicas , streamConfigDuplicateWindow + , streamConfigDenyDelete + , streamConfigAllowRollup , streamConfigAllowDirect + , streamConfigCompression , StreamConfigOption , withRetention , withStorage , withDiscard + , withDescription + , withMaxConsumers , withMaxMessages + , withMaxMessagesPerSubject , withMaxBytes , withMaxAge , withMaxMessageSize , withReplicas , withDuplicateWindow + , withDenyDelete + , withAllowRollup , withAllowDirect + , withCompression , PurgeStreamOption , withPurgeSubject , withPurgeSequence diff --git a/jetstream/JetStream/Stream/Types.hs b/jetstream/JetStream/Stream/Types.hs --- a/jetstream/JetStream/Stream/Types.hs +++ b/jetstream/JetStream/Stream/Types.hs @@ -6,6 +6,7 @@ , RetentionPolicy (..) , StorageType (..) , DiscardPolicy (..) + , StreamCompression (..) , StreamConfig (..) , StreamConfigOption , StreamConfigRequest @@ -14,13 +15,19 @@ , withRetention , withStorage , withDiscard + , withDescription + , withMaxConsumers , withMaxMessages + , withMaxMessagesPerSubject , withMaxBytes , withMaxAge , withMaxMessageSize , withReplicas , withDuplicateWindow + , withDenyDelete + , withAllowRollup , withAllowDirect + , withCompression , PurgeStreamOption , purgeStreamRequest , withPurgeSubject @@ -56,6 +63,7 @@ import qualified Data.ByteString.Base64 as Base64 import Data.Int (Int32) import Data.Maybe (catMaybes) +import qualified Data.Text as T import Data.Time.Clock (NominalDiffTime, UTCTime) import Data.Word (Word64) import JetStream.Error (JetStreamError (JetStreamDecodeError)) @@ -97,16 +105,22 @@ data StreamConfigRequest = StreamConfigRequest { streamConfigRequestName :: StreamName , streamConfigRequestSubjects :: [Subject] + , streamConfigRequestDescription :: Maybe BS.ByteString , streamConfigRequestRetention :: Maybe RetentionPolicy , streamConfigRequestStorage :: Maybe StorageType , streamConfigRequestDiscard :: Maybe DiscardPolicy + , streamConfigRequestMaxConsumers :: Maybe Int , streamConfigRequestMaxMessages :: Maybe Integer + , streamConfigRequestMaxMessagesPerSubject :: Maybe Integer , streamConfigRequestMaxBytes :: Maybe Integer , streamConfigRequestMaxAge :: Maybe NominalDiffTime , streamConfigRequestMaxMessageSize :: Maybe Int32 , streamConfigRequestReplicas :: Maybe Int , streamConfigRequestDuplicateWindow :: Maybe NominalDiffTime + , streamConfigRequestDenyDelete :: Maybe Bool + , streamConfigRequestAllowRollup :: Maybe Bool , streamConfigRequestAllowDirect :: Maybe Bool + , streamConfigRequestCompression :: Maybe StreamCompression } deriving (Eq, Show) @@ -118,26 +132,44 @@ StreamConfigRequest { streamConfigRequestName = name , streamConfigRequestSubjects = subjects + , streamConfigRequestDescription = Nothing , streamConfigRequestRetention = Nothing , streamConfigRequestStorage = Nothing , streamConfigRequestDiscard = Nothing + , streamConfigRequestMaxConsumers = Nothing , streamConfigRequestMaxMessages = Nothing + , streamConfigRequestMaxMessagesPerSubject = Nothing , streamConfigRequestMaxBytes = Nothing , streamConfigRequestMaxAge = Nothing , streamConfigRequestMaxMessageSize = Nothing , streamConfigRequestReplicas = Nothing , streamConfigRequestDuplicateWindow = Nothing + , streamConfigRequestDenyDelete = Nothing + , streamConfigRequestAllowRollup = Nothing , streamConfigRequestAllowDirect = Nothing + , streamConfigRequestCompression = Nothing } validateStreamConfigRequest :: StreamConfigRequest -> Either JetStreamError () validateStreamConfigRequest config = - case streamConfigRequestMaxMessageSize config of - Just maxMessageSize - | maxMessageSize < (-1) -> - Left (JetStreamDecodeError "stream max message size must be -1 or greater") - _ -> - Right () + validateLowerBound "stream max consumers" (streamConfigRequestMaxConsumers config) >> + validateLowerBound "stream max messages" (streamConfigRequestMaxMessages config) >> + validateLowerBound "stream max messages per subject" (streamConfigRequestMaxMessagesPerSubject config) >> + validateLowerBound "stream max bytes" (streamConfigRequestMaxBytes config) >> + case streamConfigRequestMaxMessageSize config of + Just maxMessageSize + | maxMessageSize < (-1) -> + Left (JetStreamDecodeError "stream max message size must be -1 or greater") + _ -> + Right () + where + validateLowerBound label value = + case value of + Just number + | number < (-1) -> + Left (JetStreamDecodeError (label ++ " must be -1 or greater")) + _ -> + Right () withRetention :: RetentionPolicy -> StreamConfigOption withRetention retention config = @@ -151,10 +183,22 @@ withDiscard discard config = config { streamConfigRequestDiscard = Just discard } +withDescription :: BS.ByteString -> StreamConfigOption +withDescription description config = + config { streamConfigRequestDescription = Just description } + +withMaxConsumers :: Int -> StreamConfigOption +withMaxConsumers maxConsumers config = + config { streamConfigRequestMaxConsumers = Just maxConsumers } + withMaxMessages :: Integer -> StreamConfigOption withMaxMessages maxMessages config = config { streamConfigRequestMaxMessages = Just maxMessages } +withMaxMessagesPerSubject :: Integer -> StreamConfigOption +withMaxMessagesPerSubject maxMessages config = + config { streamConfigRequestMaxMessagesPerSubject = Just maxMessages } + withMaxBytes :: Integer -> StreamConfigOption withMaxBytes maxBytes config = config { streamConfigRequestMaxBytes = Just maxBytes } @@ -177,23 +221,46 @@ withDuplicateWindow window config = config { streamConfigRequestDuplicateWindow = Just window } +withDenyDelete :: Bool -> StreamConfigOption +withDenyDelete denyDelete config = + config { streamConfigRequestDenyDelete = Just denyDelete } + +withAllowRollup :: Bool -> StreamConfigOption +withAllowRollup allowRollup config = + config { streamConfigRequestAllowRollup = Just allowRollup } + withAllowDirect :: Bool -> StreamConfigOption withAllowDirect allowDirect config = config { streamConfigRequestAllowDirect = Just allowDirect } +withCompression :: StreamCompression -> StreamConfigOption +withCompression compression config = + config { streamConfigRequestCompression = Just compression } + +data StreamCompression = NoCompression + | S2Compression + | StreamCompressionUnknown T.Text + deriving (Eq, Show) + data StreamConfig = StreamConfig - { streamConfigName :: StreamName - , streamConfigSubjects :: Maybe [Subject] - , streamConfigRetention :: RetentionPolicy - , streamConfigStorage :: StorageType - , streamConfigDiscard :: DiscardPolicy - , streamConfigMaxMessages :: Integer - , streamConfigMaxBytes :: Integer - , streamConfigMaxAge :: NominalDiffTime - , streamConfigMaxMessageSize :: Int32 - , streamConfigReplicas :: Int + { streamConfigName :: StreamName + , streamConfigSubjects :: Maybe [Subject] + , streamConfigDescription :: Maybe BS.ByteString + , streamConfigRetention :: RetentionPolicy + , streamConfigStorage :: StorageType + , streamConfigDiscard :: DiscardPolicy + , streamConfigMaxConsumers :: Int + , streamConfigMaxMessages :: Integer + , streamConfigMaxMessagesPerSubject :: Integer + , streamConfigMaxBytes :: Integer + , streamConfigMaxAge :: NominalDiffTime + , streamConfigMaxMessageSize :: Int32 + , streamConfigReplicas :: Int , streamConfigDuplicateWindow :: Maybe NominalDiffTime - , streamConfigAllowDirect :: Bool + , streamConfigDenyDelete :: Bool + , streamConfigAllowRollup :: Bool + , streamConfigAllowDirect :: Bool + , streamConfigCompression :: StreamCompression } deriving (Eq, Show) @@ -380,16 +447,22 @@ [ byteStringPair "name" (streamConfigRequestName config) , byteStringListPair "subjects" (streamConfigRequestSubjects config) ] ++ catMaybes - [ maybePair "retention" (streamConfigRequestRetention config) + [ maybeByteStringPair "description" (streamConfigRequestDescription config) + , maybePair "retention" (streamConfigRequestRetention config) , maybePair "storage" (streamConfigRequestStorage config) , maybePair "discard" (streamConfigRequestDiscard config) + , maybePair "max_consumers" (streamConfigRequestMaxConsumers config) , maybePair "max_msgs" (streamConfigRequestMaxMessages config) + , maybePair "max_msgs_per_subject" (streamConfigRequestMaxMessagesPerSubject config) , maybePair "max_bytes" (streamConfigRequestMaxBytes config) , maybeDurationPair "max_age" (streamConfigRequestMaxAge config) , maybePair "max_msg_size" (streamConfigRequestMaxMessageSize config) , maybePair "num_replicas" (streamConfigRequestReplicas config) , maybeDurationPair "duplicate_window" (streamConfigRequestDuplicateWindow config) + , maybePair "deny_delete" (streamConfigRequestDenyDelete config) + , maybePair "allow_rollup_hdrs" (streamConfigRequestAllowRollup config) , maybePair "allow_direct" (streamConfigRequestAllowDirect config) + , maybePair "compression" (streamConfigRequestCompression config) ] instance ToJSON StreamConfig where @@ -397,16 +470,22 @@ object . catMaybes $ [ Just (byteStringPair "name" (streamConfigName config)) , maybeByteStringListPair "subjects" (streamConfigSubjects config) + , maybeByteStringPair "description" (streamConfigDescription config) , Just ("retention" .= streamConfigRetention config) , Just ("storage" .= streamConfigStorage config) , Just ("discard" .= streamConfigDiscard config) + , Just ("max_consumers" .= streamConfigMaxConsumers config) , Just ("max_msgs" .= streamConfigMaxMessages config) + , Just ("max_msgs_per_subject" .= streamConfigMaxMessagesPerSubject config) , Just ("max_bytes" .= streamConfigMaxBytes config) , Just ("max_age" .= durationToNanoseconds (streamConfigMaxAge config)) , Just ("max_msg_size" .= streamConfigMaxMessageSize config) , Just ("num_replicas" .= streamConfigReplicas config) , maybeDurationPair "duplicate_window" (streamConfigDuplicateWindow config) + , Just ("deny_delete" .= streamConfigDenyDelete config) + , Just ("allow_rollup_hdrs" .= streamConfigAllowRollup config) , Just ("allow_direct" .= streamConfigAllowDirect config) + , Just ("compression" .= streamConfigCompression config) ] instance FromJSON StreamConfig where @@ -415,16 +494,37 @@ StreamConfig <$> parseByteStringField value "name" <*> parseOptionalByteStringListField value "subjects" + <*> parseOptionalByteStringField value "description" <*> value .: "retention" <*> value .: "storage" <*> value .: "discard" + <*> value .:? "max_consumers" .!= (-1) <*> value .: "max_msgs" + <*> value .:? "max_msgs_per_subject" .!= (-1) <*> value .: "max_bytes" <*> parseDurationField value "max_age" <*> value .:? "max_msg_size" .!= (-1) <*> value .: "num_replicas" <*> parseOptionalDurationField value "duplicate_window" + <*> value .:? "deny_delete" .!= False + <*> value .:? "allow_rollup_hdrs" .!= False <*> value .: "allow_direct" + <*> value .:? "compression" .!= NoCompression + +instance ToJSON StreamCompression where + toJSON compression = + String $ + case compression of + NoCompression -> "none" + S2Compression -> "s2" + StreamCompressionUnknown name -> name + +instance FromJSON StreamCompression where + parseJSON = withText "StreamCompression" $ \value -> + case value of + "none" -> pure NoCompression + "s2" -> pure S2Compression + _ -> pure (StreamCompressionUnknown value) instance ToJSON PurgeStreamRequest where toJSON request = diff --git a/natskell.cabal b/natskell.cabal --- a/natskell.cabal +++ b/natskell.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: natskell -version: 1.1.0.1 +version: 1.2.0.0 synopsis: A NATS client library written in Haskell tested-with: GHC ==8.8, @@ -63,6 +63,7 @@ Client JetStream.API JetStream.API.Consumer + JetStream.API.KeyValue JetStream.API.Management JetStream.API.Message JetStream.API.Publish @@ -89,6 +90,9 @@ JetStream.Consumer.API JetStream.Consumer.Types JetStream.Error + JetStream.KeyValue + JetStream.KeyValue.API + JetStream.KeyValue.Types JetStream.Message JetStream.Message.API JetStream.Message.Types @@ -253,6 +257,7 @@ SubscriptionStoreSpec WaitGroupSpec JetStream.ConsumerSpec + JetStream.KeyValueSpec JetStream.MessageSpec JetStream.ProtocolSpec JetStream.PublishSpec diff --git a/test/Integration/ClientSpec.hs b/test/Integration/ClientSpec.hs --- a/test/Integration/ClientSpec.hs +++ b/test/Integration/ClientSpec.hs @@ -1970,12 +1970,10 @@ ] [] putMVar fetchVar result - deleteCurrent <- capturePublish firstConn "$JS.API.CONSUMER.DELETE.PROTO_STREAM.PROTO_ORDERED_1" - replyToCapturedPublish firstConn deleteCurrent protoDeleteConsumerResponse - createNext <- capturePublish firstConn "$JS.API.CONSUMER.CREATE.PROTO_STREAM.PROTO_ORDERED_2" - replyToCapturedPublish firstConn createNext $ - protoOrderedConsumerInfoResponseFor "PROTO_ORDERED_2" - _nextRequest <- capturePublish firstConn "$JS.API.CONSUMER.MSG.NEXT.PROTO_STREAM.PROTO_ORDERED_2" + initialInfo <- capturePublish firstConn "$JS.API.CONSUMER.INFO.PROTO_STREAM.PROTO_ORDERED_1" + replyToCapturedPublish firstConn initialInfo $ + protoOrderedConsumerInfoResponseFor "PROTO_ORDERED_1" + _nextRequest <- capturePublish firstConn "$JS.API.CONSUMER.MSG.NEXT.PROTO_STREAM.PROTO_ORDERED_1" Network.Socket.close firstConn fetchResult <- timeout 1000000 (takeMVar fetchVar) case fetchResult of @@ -1992,9 +1990,9 @@ void . forkIO $ do result <- JetStream.orderedConsumerInfo ordered [] putMVar infoVar result - infoRequest <- capturePublish secondConn "$JS.API.CONSUMER.INFO.PROTO_STREAM.PROTO_ORDERED_2" + infoRequest <- capturePublish secondConn "$JS.API.CONSUMER.INFO.PROTO_STREAM.PROTO_ORDERED_1" replyToCapturedPublish secondConn infoRequest $ - protoOrderedConsumerInfoResponseFor "PROTO_ORDERED_2" + protoOrderedConsumerInfoResponseFor "PROTO_ORDERED_1" infoResult <- timeout 1000000 (takeMVar infoVar) case infoResult of Nothing -> @@ -2002,7 +2000,7 @@ Just (Left err) -> expectationFailure ("ordered consumer info failed after reconnect: " ++ show err) Just (Right info) -> - JetStream.consumerInfoName info `shouldBe` "PROTO_ORDERED_2" + JetStream.consumerInfoName info `shouldBe` "PROTO_ORDERED_1" close client [] Network.Socket.close secondConn Network.Socket.close sock diff --git a/test/PublicAPI/Main.hs b/test/PublicAPI/Main.hs --- a/test/PublicAPI/Main.hs +++ b/test/PublicAPI/Main.hs @@ -9,6 +9,7 @@ import qualified Client import qualified JetStream.API as JetStream import qualified JetStream.API.Consumer as Consumer +import qualified JetStream.API.KeyValue as KeyValue import qualified JetStream.API.Management as Management import qualified JetStream.API.Message as JetStreamMessage import qualified JetStream.API.Publish as JetStreamPublish @@ -173,6 +174,42 @@ [] [] (const (pure ())) + bucket <- KeyValue.createKeyValueBucket + (JetStream.keyValues jetStream) + "CACHE" + [ KeyValue.withKeyValueHistory 3 + , KeyValue.withKeyValueMaxValueSize 1048576 + , KeyValue.withKeyValueStorage Stream.MemoryStorage + ] + [] + _ <- KeyValue.createOrUpdateKeyValueBucket + (JetStream.keyValues jetStream) + "CACHE" + [KeyValue.withKeyValueHistory 3] + [] + case bucket of + Left _ -> pure () + Right keyValue -> do + revision <- KeyValue.putKeyValueEntry + (JetStream.keyValues jetStream) keyValue "process/one" "snapshot" [] + case revision of + Left _ -> pure () + Right current -> do + _ <- KeyValue.updateKeyValueEntry + (JetStream.keyValues jetStream) keyValue "process/one" "new" current [] + pure () + watcher <- KeyValue.watchKeyValues + (JetStream.keyValues jetStream) keyValue ["process.*"] [] [] + case watcher of + Left _ -> pure () + Right handle -> do + _ <- KeyValue.fetchKeyValueWatch + (JetStream.keyValues jetStream) handle + [JetStreamMessage.withFetchBatch 10] + [] + _ <- KeyValue.stopKeyValueWatch + (JetStream.keyValues jetStream) handle [] + pure () pure () messageOperations diff --git a/test/Unit/JetStream/KeyValueSpec.hs b/test/Unit/JetStream/KeyValueSpec.hs new file mode 100644 --- /dev/null +++ b/test/Unit/JetStream/KeyValueSpec.hs @@ -0,0 +1,151 @@ +{-# LANGUAGE OverloadedStrings #-} + +module JetStream.KeyValueSpec (spec) where + +import Data.Aeson (eitherDecode, encode, object, (.=)) +import JetStream.KeyValue.Types +import qualified JetStream.Stream.Types as Stream +import JetStream.Types + ( StorageType (FileStorage, MemoryStorage) + ) +import Test.Hspec + +spec :: Spec +spec = do + describe "key-value identifiers" $ do + it "accepts official bucket and key character sets" $ do + validateKeyValueBucketName "CACHE_v1-2" `shouldBe` Right () + validateKeyValueKey "tenant/one.process_id=value" `shouldBe` Right () + validateKeyValuePattern "tenant.*.>" `shouldBe` Right () + validateKeyValuePattern ">" `shouldBe` Right () + + it "rejects invalid buckets, keys, and search patterns" $ do + validateKeyValueBucketName "" `shouldBe` Left (KeyValueInvalidBucketName "") + validateKeyValueBucketName "cache.one" `shouldBe` Left (KeyValueInvalidBucketName "cache.one") + validateKeyValueKey ".hidden" `shouldBe` Left (KeyValueInvalidKey ".hidden") + validateKeyValueKey "trailing." `shouldBe` Left (KeyValueInvalidKey "trailing.") + validateKeyValueKey "empty..token" `shouldBe` Left (KeyValueInvalidKey "empty..token") + validateKeyValueKey "has space" `shouldBe` Left (KeyValueInvalidKey "has space") + validateKeyValuePattern "empty..token" `shouldBe` Left (KeyValueInvalidPattern "empty..token") + validateKeyValuePattern "tenant.>.*" `shouldBe` Left (KeyValueInvalidPattern "tenant.>.*") + + describe "key-value configuration" $ do + it "uses JetStream key-value defaults and normalizes zero values" $ do + let defaults = keyValueConfig "CACHE" [] + normalized = keyValueConfig "CACHE" + [ withKeyValueMaxValueSize 0 + , withKeyValueHistory 0 + , withKeyValueMaxBytes 0 + , withKeyValueReplicas 0 + ] + defaults `shouldBe` normalized + keyValueConfigMaxValueSize defaults `shouldBe` (-1) + keyValueConfigHistory defaults `shouldBe` 1 + keyValueConfigMaxBytes defaults `shouldBe` (-1) + keyValueConfigStorage defaults `shouldBe` FileStorage + keyValueConfigReplicas defaults `shouldBe` 1 + keyValueConfigTTL defaults `shouldBe` 0 + + it "validates bounded history and resource numbers" $ do + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueHistory 64]) + `shouldBe` Right () + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueHistory 65]) + `shouldBe` Left (KeyValueInvalidHistory 65) + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueMaxValueSize (-2)]) + `shouldBe` Left (KeyValueInvalidMaxValueSize (-2)) + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueMaxBytes (-2)]) + `shouldBe` Left (KeyValueInvalidMaxBytes (-2)) + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueTTL (-1)]) + `shouldBe` Left (KeyValueInvalidTTL (-1)) + validateKeyValueConfig (keyValueConfig "CACHE" [withKeyValueReplicas (-1)]) + `shouldBe` Left (KeyValueInvalidReplicas (-1)) + + it "maps configuration to an official KV backing stream" $ do + let config = keyValueConfig "CACHE" + [ withKeyValueDescription "binary snapshots" + , withKeyValueMaxValueSize 1048576 + , withKeyValueHistory 3 + , withKeyValueTTL 60 + , withKeyValueMaxBytes 10485760 + , withKeyValueStorage MemoryStorage + , withKeyValueReplicas 2 + , withKeyValueCompression True + ] + request = Stream.streamConfigRequest + (keyValueStreamName "CACHE") + [keyValuePatternSubject "CACHE" ">"] + (keyValueStreamOptions config) + eitherDecode (encode request) `shouldBe` Right (object + [ "name" .= ("KV_CACHE" :: String) + , "subjects" .= ["$KV.CACHE.>" :: String] + , "description" .= ("binary snapshots" :: String) + , "retention" .= ("limits" :: String) + , "storage" .= ("memory" :: String) + , "discard" .= ("new" :: String) + , "max_consumers" .= (-1 :: Int) + , "max_msgs" .= (-1 :: Integer) + , "max_msgs_per_subject" .= (3 :: Integer) + , "max_bytes" .= (10485760 :: Integer) + , "max_age" .= (60000000000 :: Integer) + , "max_msg_size" .= (1048576 :: Int) + , "num_replicas" .= (2 :: Int) + , "duplicate_window" .= (60000000000 :: Integer) + , "deny_delete" .= True + , "allow_rollup_hdrs" .= True + , "allow_direct" .= True + , "compression" .= ("s2" :: String) + ]) + + describe "key-value entries" $ do + it "decodes delete and purge tombstones from stored headers" $ do + keyValueEntryOperation <$> keyValueEntryFromStreamMessage bucket "deleted" (message "deleted" "DEL") + `shouldBe` Right KeyValueDelete + keyValueEntryOperation <$> keyValueEntryFromStreamMessage bucket "purged" (message "purged" "PURGE") + `shouldBe` Right KeyValuePurge + keyValueEntryOperation <$> keyValueEntryFromStreamMessage bucket "expired" (marker "expired" "MaxAge") + `shouldBe` Right KeyValuePurge + keyValueEntryOperation <$> keyValueEntryFromStreamMessage bucket "removed" (marker "removed" "Remove") + `shouldBe` Right KeyValueDelete + + it "rejects a revision whose stream subject belongs to another key" $ do + keyValueEntryFromStreamMessage bucket "expected" (message "other" "PUT") + `shouldBe` Left (KeyValueKeyNotFound "CACHE" "expected") + + it "rejects malformed stored headers" $ do + case keyValueEntryFromStreamMessage bucket "broken" (brokenHeaders "broken") of + Left (KeyValueDecodeError _) -> pure () + result -> expectationFailure ("expected header decode failure, got " ++ show result) + + describe "key-value watch options" $ do + it "rejects history combined with updates-only" $ do + validateKeyValueWatchConfig + (keyValueWatchConfig [withKeyValueIncludeHistory, withKeyValueUpdatesOnly]) + `shouldBe` Left KeyValueInvalidWatchOptions + where + bucket = KeyValueBucket "CACHE" + message key operation = + Stream.StreamMessage + { Stream.streamMessageSubject = keyValueSubject "CACHE" key + , Stream.streamMessageSequence = 10 + , Stream.streamMessageHeadersRaw = Just + ("NATS/1.0\r\nKV-Operation: " <> operation <> "\r\n\r\n") + , Stream.streamMessagePayload = Just "" + , Stream.streamMessageTime = read "2026-01-01 00:00:00 UTC" + } + marker key reason = + Stream.StreamMessage + { Stream.streamMessageSubject = keyValueSubject "CACHE" key + , Stream.streamMessageSequence = 10 + , Stream.streamMessageHeadersRaw = Just + ("NATS/1.0\r\nNats-Marker-Reason: " <> reason <> "\r\n\r\n") + , Stream.streamMessagePayload = Just "" + , Stream.streamMessageTime = read "2026-01-01 00:00:00 UTC" + } + brokenHeaders key = + Stream.StreamMessage + { Stream.streamMessageSubject = keyValueSubject "CACHE" key + , Stream.streamMessageSequence = 10 + , Stream.streamMessageHeadersRaw = Just "not-a-nats-header" + , Stream.streamMessagePayload = Just "" + , Stream.streamMessageTime = read "2026-01-01 00:00:00 UTC" + } diff --git a/test/Unit/JetStream/StreamSpec.hs b/test/Unit/JetStream/StreamSpec.hs --- a/test/Unit/JetStream/StreamSpec.hs +++ b/test/Unit/JetStream/StreamSpec.hs @@ -46,6 +46,26 @@ (streamConfigRequest "ORDERS" [] [withMaxMessageSize minBound]) `shouldBe` Left (JetStreamDecodeError "stream max message size must be -1 or greater") + it "encodes key-value backing stream controls" $ do + let request = streamConfigRequest "KV_ORDERS" ["$KV.ORDERS.>"] + [ withDescription "orders key-value bucket" + , withMaxConsumers (-1) + , withMaxMessagesPerSubject 3 + , withDenyDelete True + , withAllowRollup True + , withCompression S2Compression + ] + eitherDecode (encode request) `shouldBe` Right (object + [ "name" .= ("KV_ORDERS" :: String) + , "subjects" .= ["$KV.ORDERS.>" :: String] + , "description" .= ("orders key-value bucket" :: String) + , "max_consumers" .= (-1 :: Int) + , "max_msgs_per_subject" .= (3 :: Integer) + , "deny_delete" .= True + , "allow_rollup_hdrs" .= True + , "compression" .= ("s2" :: String) + ]) + describe "StreamConfig response JSON" $ do it "normalizes a missing max message size to unlimited" $ do fmap streamConfigMaxMessageSize (eitherDecode streamConfigWithoutMaxMessageSizeJSON) @@ -59,16 +79,22 @@ StreamConfig { streamConfigName = "ORDERS" , streamConfigSubjects = Just ["orders.>"] + , streamConfigDescription = Just "order events" , streamConfigRetention = LimitsPolicy , streamConfigStorage = MemoryStorage , streamConfigDiscard = DiscardOld + , streamConfigMaxConsumers = -1 , streamConfigMaxMessages = -1 + , streamConfigMaxMessagesPerSubject = 3 , streamConfigMaxBytes = -1 , streamConfigMaxAge = 0 , streamConfigMaxMessageSize = maxBound , streamConfigReplicas = 1 , streamConfigDuplicateWindow = Nothing + , streamConfigDenyDelete = True + , streamConfigAllowRollup = True , streamConfigAllowDirect = False + , streamConfigCompression = S2Compression } streamConfigWithoutMaxMessageSizeJSON :: LBS.ByteString