shibuya-kiroku-adapter 0.4.0.1 → 0.5.0.0

raw patch · 4 files changed

+104/−8 lines, 4 filesdep ~kiroku-storePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: kiroku-store

API changes (from Hackage documentation)

Files

• CHANGELOG.md +17/−1
• shibuya-kiroku-adapter.cabal +3/−3
• src/Shibuya/Adapter/Kiroku.hs +24/−4
• test/Main.hs +60/−0

@@ -1,6 +1,22 @@ # Changelog -## Unreleased+## 0.5.0.0 — 2026-08-11++### Breaking Changes++* `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` gain+  `missingCheckpointPolicy`. Their smart constructors default it to+  `FromBeginning`; exhaustive record literals must choose a policy or use the+  smart constructors.+* Requires `kiroku-store ^>=0.5`.++### New Features++* The adapter re-exports `MissingCheckpointPolicy` and forwards it to every+  underlying Kiroku worker. `FromCurrentHead` supports future-only Shibuya+  processors without historical side effects, while `FailIfMissing` preserves+  mandatory checkpoint provisioning. A consumer-group policy is applied+  independently to every member key; existing rows always win.  ## 0.4.0.1 — 2026-08-09 

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            shibuya-kiroku-adapter-version:         0.4.0.1+version:         0.5.0.0 synopsis:   Kiroku event store adapter for the Shibuya queue processing framework @@ -50,7 +50,7 @@     , effectful-core                         >=2.5   && <2.7     , hs-opentelemetry-api                   ^>=1.0     , hs-opentelemetry-semantic-conventions  ^>=1.40-    , kiroku-store                           ^>=0.4+    , kiroku-store                           ^>=0.5     , shibuya-core                           >=0.8   && <0.9     , stm                                    >=2.5   && <2.6     , streamly-core                          >=0.3   && <0.4@@ -78,7 +78,7 @@     , hasql-pool              >=1.2  && <1.5     , hs-opentelemetry-api    ^>=1.0     , hspec                   >=2.10 && <2.12-    , kiroku-store            ^>=0.4+    , kiroku-store            ^>=0.5     , kiroku-test-support     , lens                    >=5.2  && <5.4     , shibuya-core            >=0.8  && <0.9

@@ -132,6 +132,7 @@     SubscriptionTarget (..),     ConsumerGroup (..),     EventTypeFilter (..),+    MissingCheckpointPolicy (..), ) where  import Control.Exception (SomeException)@@ -146,6 +147,7 @@     ConsumerGroup (..),     EventTypeFilter (..),     InvalidConsumerGroup (..),+    MissingCheckpointPolicy (..),     SubscriptionConfig,     SubscriptionName (..),     SubscriptionResult (..),@@ -204,6 +206,13 @@     the underlying 'Kiroku.Store.Subscription.subscribe' call, which throws     'Kiroku.Store.Subscription.Types.InvalidConsumerGroup' on violation.     -}+    , missingCheckpointPolicy :: !MissingCheckpointPolicy+    {- ^ What the underlying Kiroku worker does when this adapter's exact+    @(subscriptionName, consumer-group member)@ checkpoint row is absent.+    'FromBeginning' is the compatibility default; use 'FromCurrentHead' for a+    future-only processor or 'FailIfMissing' when prior provisioning is+    mandatory. Existing checkpoints always win.+    -}     , eventTypeFilter :: !EventTypeFilter     {- ^ Which event types this adapter delivers. Pass 'AllEventTypes' (deliver     everything) or @'OnlyEventTypes' s@ to receive only events whose type is in@@ -230,7 +239,8 @@  {- | A 'KirokuAdapterConfig' with sensible defaults: @batchSize = 100@, @bufferSize = 256@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@-(ordinary single-consumer subscription), @eventTypeFilter = 'AllEventTypes'@+(ordinary single-consumer subscription), @missingCheckpointPolicy =+'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@ (deliver every type), and @selector = 'Nothing'@ (no extra predicate filtering). Supply the subscription name and target; override individual fields with record-update syntax.@@ -255,6 +265,7 @@         , bufferSize = 256         , queueCapacity = 16         , consumerGroup = Nothing+        , missingCheckpointPolicy = FromBeginning         , eventTypeFilter = AllEventTypes         , selector = Nothing         }@@ -303,7 +314,7 @@     KirokuStore ->     KirokuAdapterConfig ->     Eff es (Adapter es RecordedEvent)-kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, consumerGroup = cg, eventTypeFilter = etf, selector = sel} = do+kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} = do     -- Build from 'defaultSubscriptionConfig' and override only the non-default     -- fields. Using the smart constructor (rather than a full record literal)     -- means any future field added to 'SubscriptionConfigM' is inherited at its@@ -314,6 +325,7 @@                 { Sub.batchSize = bs                 , Sub.queueCapacity = qCap                 , Sub.consumerGroup = cg+                , Sub.missingCheckpointPolicy = checkpointPolicy                 , Sub.eventTypeFilter = etf                 , Sub.selector = sel                 }@@ -380,6 +392,11 @@     -}     , memberConcurrency :: !Concurrency     -- ^ Per-member concurrency; must be 'Serial' (validated).+    , missingCheckpointPolicy :: !MissingCheckpointPolicy+    {- ^ Missing-checkpoint policy applied independently to every member key.+    Existing member rows always win; a new 'FromCurrentHead' group seeds every+    member at the head each member observes during startup.+    -}     , eventTypeFilter :: !EventTypeFilter     {- ^ Event-type filter applied to /every/ member (the same filter on each).     'AllEventTypes' delivers everything; @'OnlyEventTypes' s@ delivers only the@@ -404,7 +421,8 @@  {- | A 'KirokuConsumerGroupConfig' with sensible defaults: @memberConcurrency = 'Serial'@ (the only legal per-member concurrency), @batchSize = 100@,-@bufferSize = 256@, @queueCapacity = 16@, @eventTypeFilter = 'AllEventTypes'@+@bufferSize = 256@, @queueCapacity = 16@, @missingCheckpointPolicy =+'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@ (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering). Supply the subscription name, target, and group size. -}@@ -419,6 +437,7 @@         , bufferSize = 256         , queueCapacity = 16         , memberConcurrency = Serial+        , missingCheckpointPolicy = FromBeginning         , eventTypeFilter = AllEventTypes         , selector = Nothing         }@@ -466,7 +485,7 @@     KirokuConsumerGroupConfig ->     Handler es RecordedEvent ->     Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])-kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig{subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, eventTypeFilter = etf, selector = sel} handler =+kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig{subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler =     kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler   where     mkMemberAdapter m =@@ -479,6 +498,7 @@                 , bufferSize = buf                 , queueCapacity = qCap                 , consumerGroup = Just (ConsumerGroup{member = m, size = n})+                , missingCheckpointPolicy = checkpointPolicy                 , eventTypeFilter = etf                 , selector = sel                 }

@@ -304,6 +304,51 @@                 collected <- readIORef ref                 length collected `shouldBe` 10 +            it "supports a future-only FromCurrentHead adapter" $ \store -> do+                Right _ <-+                    runStoreIO store $+                        appendToStream+                            (StreamName "shibuya-future-only-events")+                            NoStream+                            (map (\i -> makeEvent ("Historical" <> T.pack (show i)) (Aeson.object [])) [1 .. 4 :: Int])+                ref <- newIORef ([] :: [RecordedEvent])+                countVar <- newTVarIO (0 :: Int)++                runEff $ runTracingNoop $ do+                    adapter <-+                        kirokuAdapter store $+                            defaultKirokuAdapterConfig (SubscriptionName "shibuya-future-only") AllStreams+                                & #missingCheckpointPolicy .~ FromCurrentHead+                    liftIO $+                        waitForCheckpointPosition+                            store+                            (SubscriptionCheckpointKey (SubscriptionName "shibuya-future-only") 0)+                            (GlobalPosition 4)+                    liftIO $ do+                        Right _ <-+                            runStoreIO store $+                                appendToStream+                                    (StreamName "shibuya-future-only-events")+                                    StreamExists+                                    [makeEvent "Future" (Aeson.object [])]+                        pure ()+                    let handler ingested = do+                            liftIO $ do+                                modifyIORef' ref (envelopePayload ingested :)+                                atomically $ do+                                    count <- readTVar countVar+                                    writeTVar countVar (count + 1)+                            pure AckOk+                    result <- runApp defaultAppConfig [(ProcessorId "future-only", mkProcessor adapter handler)]+                    case result of+                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                        Right appHandle -> do+                            liftIO $ waitForCount countVar 1 10_000_000+                            stopApp appHandle++                collected <- reverse <$> readIORef ref+                map globalPos collected `shouldBe` [5]+             it "delivers live events through Shibuya pipeline" $ \store -> do                 ref <- newIORef ([] :: [RecordedEvent])                 countVar <- newTVarIO (0 :: Int)@@ -1037,6 +1082,21 @@     case result of         Nothing -> fail ("Timed out waiting for " <> label)         Just a -> pure a++waitForCheckpointPosition :: KirokuStore -> SubscriptionCheckpointKey -> GlobalPosition -> IO ()+waitForCheckpointPosition store expectedKey expectedPosition =+    within "subscription checkpoint initialization" loop+  where+    loop = do+        Right (SubscriptionCheckpointInventory _ checkpoints) <-+            runStoreIO store subscriptionCheckpointInventory+        let positions =+                [ (SubscriptionCheckpointKey name member, position)+                | SubscriptionCheckpoint name member position _ <- toList checkpoints+                ]+        if (expectedKey, expectedPosition) `elem` positions+            then pure ()+            else threadDelay 20_000 >> loop  makeEvent :: Text -> Value -> EventData makeEvent typ p =