natskell 1.1.0.1 → 1.2.0.0

raw patch · 16 files changed

+1841/−34 lines, 16 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• client/JetStream/API.hs +4/−0
• client/JetStream/API/KeyValue.hs +4/−0
• client/JetStream/Client.hs +9/−4
• internal/Plumbing/Parser/Attoparsec.hs +5/−0
• jetstream/JetStream/KeyValue.hs +811/−0
• jetstream/JetStream/KeyValue/API.hs +81/−0
• jetstream/JetStream/KeyValue/Types.hs +544/−0
• jetstream/JetStream/Message.hs +23/−2
• jetstream/JetStream/Options.hs +2/−0
• jetstream/JetStream/Stream/API.hs +13/−0
• jetstream/JetStream/Stream/Types.hs +118/−18
• natskell.cabal +6/−1
• test/Integration/ClientSpec.hs +7/−9
• test/PublicAPI/Main.hs +37/−0
• test/Unit/JetStream/KeyValueSpec.hs +151/−0
• test/Unit/JetStream/StreamSpec.hs +26/−0

@@ -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

@@ -0,0 +1,4 @@+-- | JetStream key-value contract.+module JetStream.API.KeyValue (module JetStream.KeyValue.API) where++import           JetStream.KeyValue.API

@@ -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     }

@@ -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 =

@@ -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)

@@ -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

@@ -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 }

@@ -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

@@ -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

@@ -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

@@ -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 =

@@ -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

@@ -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

@@ -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

@@ -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"+        }

@@ -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