StreamHead.hs

module Test.StreamHead (spec) where

import Control.Concurrent.Async qualified as Async
import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
import Control.Lens ((&), (.~), (^.))
import Control.Monad (forM_, replicateM_, void)
import Data.Aeson (Value (Null))
import Data.Generics.Labels ()
import Data.IORef (modifyIORef', newIORef, readIORef)
import Data.Text (Text)
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
import System.Timeout (timeout)
import Test.Helpers (makeEvent, withTestStore, withTestStoreSettings)
import Test.Hspec

spec :: Spec
spec = describe "stream head" $ do
  forM_ [False, True] $ \resource -> describe (if resource then "resource runner" else "direct runner") $ do
    let readHead store name =
          if resource
            then runEff . runErrorNoCallStack @StoreError . runKirokuStoreWith store . runStoreResource $ getStreamWithHead name
            else runStoreIO store (getStreamWithHead name)
        check store name version headPosition = do
          Right (Just metadata) <- runStoreIO store (getStream name)
          metadata ^. #version `shouldBe` StreamVersion version
          readHead store name `shouldReturn` Right (Just (metadata, headPosition))
    it "distinguishes absent, empty and reserved $all in empty and populated stores" $ withTestStore $ \store -> do
      readHead store (StreamName "missing") `shouldReturn` Right Nothing
      raw store "INSERT INTO streams(stream_name) VALUES ('empty')"
      check store (StreamName "empty") 0 Nothing
      check store (StreamName "$all") 0 Nothing
      appendOne store (StreamName "origin")
      check store (StreamName "$all") 1 Nothing
    it "captures interleaved heads, logical lifecycle and a fresh identity after hard deletion" $ withTestStore $ \store -> do
      let a = StreamName "a"; b = StreamName "b"
      mapM_ (appendOne store) [a, b, a, b]
      originated <- appendPosition store a
      originated `shouldBe` GlobalPosition 5
      check store a 3 (Just originated)
      appendOne store b
      check store a 3 (Just originated)
      Right (Just _) <- runStoreIO store (setStreamTruncateBefore a (StreamVersion 3))
      check store a 3 (Just originated)
      Right (Just _) <- runStoreIO store (softDeleteStream a)
      check store a 3 (Just originated)
      Right (Just (old, _)) <- readHead store a
      Right (Just _) <- runStoreIO store (hardDeleteStream a)
      readHead store a `shouldReturn` Right Nothing
      raw store "INSERT INTO streams(stream_name) VALUES ('a')"
      check store a 0 Nothing
      Right (Just (fresh, _)) <- readHead store a
      fresh ^. #id `shouldNotBe` (old ^. #id)
      recreated <- appendPosition store a
      recreated `shouldBe` GlobalPosition 7
      check store a 1 (Just recreated)
    it "ignores links while tracking subsequent originated appends" $ withTestStore $ \store -> do
      let mixed = StreamName "mixed"; source = StreamName "source"; linked = StreamName "linked"
      appendOne store mixed
      appendOne store source
      Right events <- runStoreIO store (readAllForward (GlobalPosition 0) 10)
      let eid = (events V.! 1) ^. #eventId
      Right _ <- runStoreIO store (linkToStream linked [eid])
      check store linked 1 Nothing
      Right _ <- runStoreIO store (linkToStream mixed [eid])
      check store mixed 2 (Just ((events V.! 0) ^. #globalPosition))
      next <- appendPosition store mixed
      next `shouldBe` GlobalPosition 3
      check store mixed 3 (Just next)
  it "never invokes the event decode hook" $ do
    calls <- newIORef (0 :: Int)
    let hook _ = modifyIORef' calls (+ 1) >> error "unexpected event decode"
    withTestStoreSettings (\settings -> settings & #storeSettings .~ defaultStoreSettings {decodeHook = Just hook}) $ \store -> do
      expected <- appendPosition store (StreamName "hook")
      Right (Just (_, actual)) <- runStoreIO store (getStreamWithHead (StreamName "hook"))
      actual `shouldBe` Just expected
      readIORef calls `shouldReturn` 0
  it "observes version and head from one snapshot while appends commit" $ withTestStore $ \store -> do
    let name = StreamName "concurrent"
    appendOne store name
    start <- newEmptyMVar
    let writer = takeMVar start >> replicateM_ 100 (appendOne store name)
        reader = replicateM_ 200 $ do
          Right (Just (info, Just (GlobalPosition position))) <- runStoreIO store (getStreamWithHead name)
          info ^. #version `shouldBe` StreamVersion position
    outcome <- timeout 20_000_000 $ Async.withAsync writer $ \worker -> do
      putMVar start ()
      reader
      Async.wait worker
    outcome `shouldBe` Just ()
    Right (Just (info, headPosition)) <- runStoreIO store (getStreamWithHead name)
    info ^. #version `shouldBe` StreamVersion 101
    headPosition `shouldBe` Just (GlobalPosition 101)

appendOne :: KirokuStore -> StreamName -> IO ()
appendOne store name = void (appendPosition store name)

appendPosition :: KirokuStore -> StreamName -> IO GlobalPosition
appendPosition store name = do
  result <- runStoreIO store (appendToStream name AnyVersion [makeEvent "StreamHeadEvent" Null])
  either (fail . show) (pure . (^. #globalPosition)) result

raw :: KirokuStore -> Text -> IO ()
raw store command = Pool.use (store ^. #pool) (Session.script command) >>= either (expectationFailure . show) pure