TruncateBefore.hs

-- | Tests for the logical truncate-before marker (close-the-book
-- compaction, ExecPlan docs/plans/65).
--
-- The marker is a per-stream cursor that hides a prefix from the /ordered
-- per-stream/ reads ('readStreamForward' / 'readStreamBackward' and the paged
-- 'readStreamForwardStream') while leaving the global @$all@ log, category
-- reads, and the physical row count untouched. The distinguishing property —
-- and the single most important assertion here — is that after a truncate the
-- @$all@ log and 'readCategory' still return the full history and
-- 'countEvents' is unchanged: nothing is deleted, only hidden, and the
-- operation is fully reversible.
module Test.TruncateBefore (spec) where

import Control.Lens ((^.))
import Data.Aeson qualified as Aeson
import Data.Generics.Labels ()
import Data.Text qualified as T
import Data.Vector qualified as V
import Kiroku.Store
import Streamly.Data.Stream qualified as Stream
import Test.Helpers (countEvents, makeEvent, withTestStore)
import Test.Hspec

-- | Append @n@ events @E1 .. En@ to a fresh stream and return its name.
seedStream :: KirokuStore -> StreamName -> Int -> IO ()
seedStream store name n = do
  let events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. n]]
  Right _ <- runStoreIO store $ appendToStream name NoStream events
  pure ()

-- | The per-stream versions returned by a forward read, in order.
forwardVersions :: KirokuStore -> StreamName -> IO [StreamVersion]
forwardVersions store name = do
  Right v <- runStoreIO store $ readStreamForward name (StreamVersion 0) 100
  pure (map (^. #streamVersion) (V.toList v))

spec :: Spec
spec = describe "TruncateBefore" $
  around withTestStore $ do
    it "bounds per-stream reads to the kept suffix" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6 -- versions 1..6; treat v6 as the snapshot
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)

      forwardVersions store name `shouldReturn` [StreamVersion 6]

      Right back <- runStoreIO store $ readStreamBackward name (StreamVersion 0) 100
      map (^. #streamVersion) (V.toList back) `shouldBe` [StreamVersion 6]

    it "leaves the $all global log and category reads intact" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6
      countBefore <- countEvents store
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)

      -- \$all global log still returns the full history.
      Right allEvents <- runStoreIO store $ readAllForward (GlobalPosition 0) 1000
      length (V.toList allEvents) `shouldBe` 6

      -- Category read still returns the full history.
      Right catEvents <- runStoreIO store $ readCategory (CategoryName "preference") (GlobalPosition 0) 1000
      length (V.toList catEvents) `shouldBe` 6

      -- Nothing was physically deleted.
      countAfter <- countEvents store
      countAfter `shouldBe` countBefore

    it "is reversible via clearStreamTruncateBefore" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)
      forwardVersions store name `shouldReturn` [StreamVersion 6]

      Right (Just _) <- runStoreIO store $ clearStreamTruncateBefore name
      forwardVersions store name
        `shouldReturn` map StreamVersion [1 .. 6]

    it "is idempotent" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)
      first <- forwardVersions store name
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)
      second <- forwardVersions store name
      first `shouldBe` [StreamVersion 4, StreamVersion 5, StreamVersion 6]
      second `shouldBe` first

    it "applies across paged reads" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)
      -- Page size 2 forces multiple pages across the kept suffix.
      Right streamed <-
        runStoreIO store $
          Stream.toList $
            fmap (^. #streamVersion) (readStreamForwardStream name (StreamVersion 0) 2)
      streamed `shouldBe` [StreamVersion 4, StreamVersion 5, StreamVersion 6]

    it "reflects the marker in getStream" $ \store -> do
      let name = StreamName "preference-abc"
      seedStream store name 6
      Right (Just info0) <- runStoreIO store $ getStream name
      (info0 ^. #truncateBefore) `shouldBe` StreamVersion 0
      Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 3)
      Right (Just info1) <- runStoreIO store $ getStream name
      (info1 ^. #truncateBefore) `shouldBe` StreamVersion 3

    it "returns Nothing for missing or soft-deleted streams" $ \store -> do
      Right missing <- runStoreIO store $ setStreamTruncateBefore (StreamName "does-not-exist") (StreamVersion 1)
      missing `shouldBe` Nothing

      let name = StreamName "preference-abc"
      seedStream store name 3
      Right (Just _) <- runStoreIO store $ softDeleteStream name
      Right deleted <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 1)
      deleted `shouldBe` Nothing

    it "rejects $all with ReservedStreamName" $ \store -> do
      result <- runStoreIO store $ setStreamTruncateBefore (StreamName "$all") (StreamVersion 1)
      case result of
        Left (ReservedStreamName (StreamName "$all")) -> pure ()
        other ->
          expectationFailure
            ("Expected Left (ReservedStreamName \"$all\"), got: " <> show other)