IntegrationSpec.hs

{-# LANGUAGE OverloadedLabels #-}

-- | Integration test: wire the collector into a real ephemeral
-- PostgreSQL-backed store, run a live @$all@ subscription, append events, and
-- assert the snapshot reflects real store activity (not just scripted inputs).
--
-- The ordering wrinkle is that 'newKirokuMetrics' needs the 'KirokuStore' but the
-- callbacks must be installed on 'ConnectionSettings' before 'withStore' creates
-- it. We resolve it with a @TVar (Maybe KirokuStore)@ the snapshot-time STM
-- readers consult; it is filled inside 'withStore'. (A 'TVar', not an 'IORef',
-- because the readers are 'STM'.)
module Test.IntegrationSpec (spec) where

import Control.Concurrent.STM
  ( STM,
    TVar,
    atomically,
    check,
    newTVarIO,
    orElse,
    readTVar,
    registerDelay,
    writeTVar,
  )
import Control.Lens ((&), (.~))
import Control.Monad (unless)
import Data.Aeson qualified as Aeson
import Data.Generics.Labels ()
import Data.IntMap.Strict qualified as IntMap
import Data.Map.Strict qualified as Map
import Data.Text qualified as T
import Test.Hspec

import Kiroku.Metrics
  ( LifecycleCounters (..),
    MetricsSnapshot (..),
    StoreGauges (..),
    SubscriptionMetrics (..),
    metricsEventHandler,
    metricsObservationHandler,
    newKirokuMetricsWith,
    snapshotMetrics,
  )
import Kiroku.Store
  ( EventData (..),
    EventType (..),
    ExpectedVersion (..),
    KirokuStore (..),
    StreamName (..),
    SubscriptionResult (..),
    SubscriptionTarget (..),
    appendToStream,
    cancel,
    defaultConnectionSettings,
    defaultSubscriptionConfig,
    runStoreIO,
    subscribe,
    wait,
    withStore,
  )
import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition)
import Kiroku.Store.Subscription.Types (SubscriptionName (..))
import Kiroku.Store.Types (GlobalPosition (..))
import Kiroku.Test.Postgres (withMigratedTestDatabase)

subName :: SubscriptionName
subName = SubscriptionName "metrics-it"

spec :: Spec
spec = describe "Kiroku.Metrics.Collector (integration)" $ do
  it "observes real store activity: appends, a running subscription, and lag" $
    withMigratedTestDatabase $ \connStr -> do
      -- Deferred store handle for the snapshot-time STM readers.
      storeVar <- newTVarIO Nothing
      km <-
        newKirokuMetricsWith
          (readPosition storeVar)
          (readSubscribers storeVar)
      let settings =
            defaultConnectionSettings connStr
              & #eventHandler .~ Just (metricsEventHandler km Nothing)
              & #observationHandler .~ Just (metricsObservationHandler km Nothing)
      withStore settings $ \store -> do
        atomically (writeTVar storeVar (Just store))

        -- Subscribe to $all on an empty store: the worker goes live
        -- immediately, so the 5 events we append next flow through the
        -- live publisher path and advance publisherPosition.
        delivered <- newTVarIO (0 :: Int)
        let cfg =
              defaultSubscriptionConfig subName AllStreams $ \_event -> do
                atomically (modifyCount delivered)
                pure Continue
        handle <- subscribe store cfg

        appendStoreEvents store (StreamName "metrics-it-stream") 5
        waitForCount delivered 5 10_000_000

        cancel handle
        _ <- wait handle

        snap <- snapshotMetrics km
        snap.store.globalPosition `shouldSatisfy` (>= 5)
        snap.counters.subscriptionsStarted `shouldBe` 1
        snap.counters.subscriptionsStoppedCancelled `shouldBe` 1
        snap.counters.eventsDelivered `shouldSatisfy` (>= 5)
        case Map.lookup "metrics-it" snap.subscriptions of
          Nothing -> expectationFailure "expected subscription \"metrics-it\" in snapshot"
          Just m -> m.lag `shouldSatisfy` (>= 0)

-- | Snapshot-time reader for the store global position, defaulting to 0 until
-- the store is filled in.
readPosition :: TVar (Maybe KirokuStore) -> STM GlobalPosition
readPosition storeVar =
  readTVar storeVar >>= maybe (pure (GlobalPosition 0)) (publisherPosition . (.publisher))

-- | Snapshot-time reader for the active-subscriber count.
readSubscribers :: TVar (Maybe KirokuStore) -> STM Int
readSubscribers storeVar =
  readTVar storeVar
    >>= maybe (pure 0) (\s -> IntMap.size <$> readTVar (subscribers s.publisher))

modifyCount :: TVar Int -> STM ()
modifyCount v = readTVar v >>= \c -> writeTVar v (c + 1)

-- | Append @n@ trivial events to a fresh stream via the IO store interpreter.
appendStoreEvents :: KirokuStore -> StreamName -> Int -> IO ()
appendStoreEvents store stream n = do
  let events =
        [ EventData
            { eventId = Nothing,
              eventType = EventType ("E" <> T.pack (show i)),
              payload = Aeson.Null,
              metadata = Nothing,
              causationId = Nothing,
              correlationId = Nothing
            }
        | i <- [1 .. n]
        ]
  result <- runStoreIO store (appendToStream stream NoStream events)
  case result of
    Right _ -> pure ()
    Left err -> expectationFailure ("appendStoreEvents failed: " <> show err)

-- | Wait until the counter reaches @target@ or the timeout (micros) fires.
waitForCount :: TVar Int -> Int -> Int -> IO ()
waitForCount countVar target timeoutMicros = do
  timeoutVar <- registerDelay timeoutMicros
  ok <-
    atomically $
      (do c <- readTVar countVar; check (c >= target); pure True)
        `orElse` (do t <- readTVar timeoutVar; check t; pure False)
  unless ok $ do
    actual <- atomically (readTVar countVar)
    expectationFailure ("Timed out waiting for " <> show target <> ", got " <> show actual)