SubscriptionDeadLetters.hs

{-# LANGUAGE OverloadedLabels #-}

module Test.SubscriptionDeadLetters (spec) where

import Control.Lens ((&), (.~), (^.))
import Control.Monad (forM_, void)
import Data.Aeson (object, (.=))
import Data.Generics.Labels ()
import Data.Int (Int32, Int64)
import Data.Text qualified as T
import Data.Time.Clock (getCurrentTime)
import Data.Vector qualified as V
import Effectful (runEff)
import Effectful.Error.Static (runErrorNoCallStack)
import Hasql.Pool qualified as Pool
import Hasql.Session qualified as Session
import Kiroku.Store hiding (HistoryRetentionInventoryQuery (..), member)
import Kiroku.Store.SQL qualified as SQL
import Test.Helpers
import Test.Hspec

spec :: Spec
spec = describe "SubscriptionDeadLetters" $ do
    it "validates page sizes including both boundaries" $ do
        forM_ [0, -1, 1001] $ \n -> mkSubscriptionDeadLetterLimit n `shouldBe` Left (SubscriptionDeadLetterLimitOutOfRange n)
        forM_ [1, 1000] $ \n -> fmap subscriptionDeadLetterLimitValue (mkSubscriptionDeadLetterLimit n) `shouldBe` Right n
        subscriptionDeadLetterLimitValue defaultSubscriptionDeadLetterLimit `shouldBe` 100
    around withTestStore $ do
        it "returns empty pages through both Store interpreters" $ \store -> do
            let query = defaultSubscriptionDeadLetterQuery (SubscriptionName "absent")
            runStoreIO store (subscriptionDeadLetters query) `shouldReturn` Right (SubscriptionDeadLetterPage V.empty Nothing)
            (runEff . runErrorNoCallStack @StoreError . runKirokuStoreWith store . runStoreResource $ subscriptionDeadLetters query)
                `shouldReturn` Right (SubscriptionDeadLetterPage V.empty Nothing)
        it "pages without omissions, isolates names and matches the internal member read" $ \store -> do
            events <- seedEvents store 7
            mapM_ (insertDeadLetterForEvent store "page") (take 5 events)
            mapM_ (insertDeadLetterForEvent store "other") (drop 5 events)
            let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "page")){limit = validLimit 2}
            first <- page store query
            second <- page store query{after = first ^. #nextCursor}
            third <- page store query{after = second ^. #nextCursor}
            map positions [first, second, third] `shouldBe` [[5, 4], [3, 2], [1]]
            third ^. #nextCursor `shouldBe` Nothing
            Right internal <- Pool.use (store ^. #pool) (Session.statement ("page", 0) SQL.readDeadLettersStmt)
            map SQL.deadLetterGlobalPosition (V.toList internal) `shouldBe` [5, 4, 3, 2, 1]
            forM_ [5, 1000] $ \n -> do
                final <- page store query{limit = validLimit n}
                positions final `shouldBe` [5, 4, 3, 2, 1]
                final ^. #nextCursor `shouldBe` Nothing
        it "merges historical members and resolves same-position ties with an exclusive cursor" $ \store -> do
            [event] <- seedEvents store 1
            forM_ [0, 1, 9] $ \member -> insertDeadLetterWith store "ties" member (object []) "tie" 2 event
            let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "ties")){limit = validLimit 1}
            first <- page store query
            second <- page store query{after = first ^. #nextCursor}
            third <- page store query{after = second ^. #nextCursor}
            map (\p -> map (\row -> row ^. #consumerGroupMember) (V.toList (p ^. #deadLetters))) [first, second, third] `shouldBe` [[9], [1], [0]]
            third ^. #nextCursor `shouldBe` Nothing
            forM_ [0, 1, 9, 7] $ \member -> do
                selected <- page store (query & #consumerGroupMember .~ Just member)
                positions selected `shouldBe` if member == 7 then [] else [1]
        it "preserves arbitrary and decode-failure JSON unchanged" $ \store -> do
            events <- seedEvents store 2
            let reasons = [object ["kind" .= ("other" :: T.Text), "detail" .= object ["code" .= (42 :: Int)]], object ["kind" .= ("decode_failure" :: T.Text), "detail" .= ("bad version" :: T.Text)]]
            forM_ (zip events reasons) $ \(event, reason) -> insertDeadLetterWith store "json" 0 reason "summary" 3 event
            result <- page store (defaultSubscriptionDeadLetterQuery (SubscriptionName "json"))
            map (\row -> row ^. #reason) (V.toList (result ^. #deadLetters)) `shouldBe` reverse reasons
            now <- getCurrentTime
            forM_ (V.toList (result ^. #deadLetters)) $ \row -> do
                row ^. #attemptCount `shouldBe` 3
                row ^. #createdAt `shouldSatisfy` (<= now)
        it "never invokes a configured event decode hook for dead-letter reads" $ \_ ->
            withTestStoreSettings (\settings -> (settings & #storeSettings .~ defaultStoreSettings{decodeHook = Just (\_ -> fail "dead-letter read invoked decode hook")})) $ \store -> do
                Right _ <- runStoreIO store $ appendToStream (StreamName "hook-1") NoStream [makeEvent "E" (object [])]
                Right () <- Pool.use (store ^. #pool) (Session.script "INSERT INTO dead_letters(subscription_name,global_position,event_id,reason,reason_summary,attempt_count) SELECT 'hook',1,event_id,'{}','fixture',1 FROM events")
                result <- page store (defaultSubscriptionDeadLetterQuery (SubscriptionName "hook"))
                positions result `shouldBe` [1]

        it "keeps a cursor valid after its event and row are hard-deleted" $ \store -> do
            events <- seedEvents store 3
            mapM_ (insertDeadLetterForEvent store "delete") events
            let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "delete")){limit = validLimit 1}
            first <- page store query
            Right _ <- runStoreIO store $ hardDeleteStream (StreamName "dead-3")
            older <- page store query{after = first ^. #nextCursor, limit = validLimit 10}
            positions older `shouldBe` [2, 1]
        it "includes legal maxBound pairs on first pages" $ \store -> do
            [event] <- seedEvents store 1
            insertDeadLetterForEvent store "boundary" event
            Right () <- Pool.use (store ^. #pool) (Session.script "UPDATE dead_letters SET global_position=9223372036854775807, dead_letter_id=9223372036854775807 WHERE subscription_name='boundary'")
            let query = defaultSubscriptionDeadLetterQuery (SubscriptionName "boundary")
            forM_ [Nothing, Just 0] $ \member -> do
                result <- page store (query & #consumerGroupMember .~ member)
                positions result `shouldBe` [maxBound]
                older <- page store (query & #consumerGroupMember .~ member & #after .~ Just (SubscriptionDeadLetterCursor (GlobalPosition maxBound) maxBound))
                positions older `shouldBe` []
        it "reads a real worker-produced poison reason without changing its checkpoint" $ \store -> do
            events <- seedEvents store 3
            waitForPublisher store (GlobalPosition 3)
            handle <- subscribe store $ defaultSubscriptionConfig (SubscriptionName "worker") AllStreams $ \event -> pure $ case event ^. #globalPosition of
                GlobalPosition 2 -> DeadLetter (DeadLetterPoison "boom")
                GlobalPosition 3 -> Stop
                _ -> Continue
            waitWithTimeout 10_000_000 handle >>= \case
                Right (Right ()) -> pure ()
                other -> expectationFailure (show other)
            result <- page store (defaultSubscriptionDeadLetterQuery (SubscriptionName "worker"))
            case V.toList (result ^. #deadLetters) of
                [row] -> do
                    row ^. #eventId `shouldBe` (events !! 1) ^. #eventId
                    row ^. #reason `shouldBe` object ["kind" .= ("poison" :: T.Text), "detail" .= ("boom" :: T.Text)]
                    row ^. #reasonSummary `shouldBe` "poison: boom"
                _ -> expectationFailure (show result)
            Pool.use (store ^. #pool) (Session.statement ("worker", 0) SQL.getCheckpointMemberStmt) `shouldReturn` Right (Just 3)

validLimit :: Int32 -> SubscriptionDeadLetterLimit
validLimit = either (error . show) (\value -> value) . mkSubscriptionDeadLetterLimit
page :: KirokuStore -> SubscriptionDeadLetterQuery -> IO SubscriptionDeadLetterPage
page store query = runStoreIO store (subscriptionDeadLetters query) >>= either (fail . show) pure
positions :: SubscriptionDeadLetterPage -> [Int64]
positions result = map (\row -> let GlobalPosition p = row ^. #globalPosition in p) (V.toList (result ^. #deadLetters))
seedEvents :: KirokuStore -> Int -> IO [RecordedEvent]
seedEvents store count = do
    forM_ [1 .. count] $ \n -> void $ runStoreIO store $ appendToStream (StreamName ("dead-" <> T.pack (show n))) NoStream [makeEvent "E" (object [])]
    Right events <- runStoreIO store $ readAllForward (GlobalPosition 0) (fromIntegral count)
    pure (V.toList events)