SubscriptionsSpec.hs

-- | EP-5 tests for the @GET /subscriptions@ endpoint and the CLI remote client.
--
--   * Cross-package shape: rows produced by the server-side mapping
--     ('subscriptionStatusRows') encode and decode back through the shared
--     @kiroku-cli@ codec — guarding against either side re-encoding locally.
--   * End-to-end: boot a real store with a live subscription, serve
--     @\/subscriptions@ from 'storeSubscriptionStatus', and assert the real CLI
--     client ('fetchRemoteSubscriptionStatusRows') reports the live subscription
--     with a sane phase and the expected cursor. A raw GET checks the by-name route
--     (known → one row, unknown → @[]@), and a server with no provider yields the
--     configured-404.
module Test.SubscriptionsSpec (spec) where

import Control.Concurrent (threadDelay)
import Data.Aeson qualified as Aeson
import Data.Int (Int32, Int64)
import Data.Map.Strict qualified as Map
import Data.Text (Text)
import Data.Text qualified as T
import Network.HTTP.Client
  ( defaultManagerSettings,
    httpLbs,
    newManager,
    parseRequest,
    responseBody,
    responseStatus,
  )
import Network.HTTP.Types.Status (statusCode)
import Network.Wai.Handler.Warp qualified as Warp
import Test.Hspec

import Kiroku.Cli.Command (RemoteEndpoint (..))
import Kiroku.Cli.Subscription.Status
  ( SubscriptionStatusRow (..),
    fetchRemoteSubscriptionStatusRows,
    subscriptionStatusRows,
  )
import Kiroku.Metrics
  ( MetricsServer (..),
    defaultConfig,
    newKirokuMetrics,
    startMetricsServer,
    stopMetricsServer,
    storeSubscriptionStatus,
    subscriptionsApp,
    withMetricsServerSubscriptions,
  )
import Kiroku.Metrics.Config (MetricsServerConfig (..))
import Kiroku.Store
  ( EventData (..),
    EventType (..),
    ExpectedVersion (..),
    KirokuStore,
    StreamName (..),
    SubscriptionResult (..),
    SubscriptionTarget (..),
    appendToStream,
    cancel,
    defaultConnectionSettings,
    defaultSubscriptionConfig,
    runStoreIO,
    subscribe,
    withStore,
  )
import Kiroku.Store.Subscription (SubscriptionStateView (..), subscriptionStates)
import Kiroku.Store.Subscription.Fsm (SubscriptionState (..))
import Kiroku.Store.Subscription.Types (SubscriptionName (..))
import Kiroku.Store.Types (GlobalPosition (..))
import Kiroku.Test.Postgres (withMigratedTestDatabase)

spec :: Spec
spec = describe "Kiroku.Metrics.Subscriptions (/subscriptions)" $ do
  it "encodes server-side rows that decode back through the shared CLI codec" $ do
    let views =
          Map.fromList
            [ ((SubscriptionName "beta", 1), mkView "beta" 1 "live" 42),
              ((SubscriptionName "alpha", 0), mkView "alpha" 0 "catching_up" 7)
            ]
        rows = subscriptionStatusRows views
    Aeson.eitherDecode (Aeson.encode rows) `shouldBe` Right rows

  it "reports a live subscription to the CLI remote client end to end" $
    withMigratedTestDatabase $ \connStr ->
      withStore (defaultConnectionSettings connStr) $ \store -> do
        m <- newKirokuMetrics store
        let name = SubscriptionName "subs-endpoint-it"
        Right _ <-
          runStoreIO store $
            appendToStream (StreamName "subs-endpoint-1") NoStream [ev "A", ev "B", ev "C"]
        handle <- subscribe store (defaultSubscriptionConfig name AllStreams (\_ -> pure Continue))
        live <- waitUntilPhase 10_000_000 store (name, 0) "live"
        live `shouldBe` True

        withMetricsServerSubscriptions (defaultConfig {port = 0}) m [] (storeSubscriptionStatus store) $ \srv -> do
          threadDelay 200_000
          let base = "http://127.0.0.1:" <> show srv.serverPort

          -- The real CLI client hits GET /subscriptions and decodes the rows.
          rowsResult <- fetchRemoteSubscriptionStatusRows (RemoteEndpoint (T.pack base))
          case rowsResult of
            Left err -> expectationFailure ("expected rows, got error: " <> T.unpack err)
            Right rows ->
              case filter (\r -> r.subscription == "subs-endpoint-it") rows of
                [row] -> do
                  row.phase `shouldBe` "live"
                  row.globalPosition `shouldBe` 3
                other -> expectationFailure ("expected exactly one matching row, got " <> show other)

          -- By-name route: known name → one row, unknown name → [].
          known <- getRows (base <> "/subscriptions/subs-endpoint-it")
          map (.subscription) known `shouldBe` ["subs-endpoint-it"]
          unknown <- getRows (base <> "/subscriptions/does-not-exist")
          unknown `shouldBe` []

        cancel handle

  it "returns a configured-404 when no provider is wired" $
    withMigratedTestDatabase $ \connStr ->
      withStore (defaultConnectionSettings connStr) $ \store -> do
        m <- newKirokuMetrics store
        srv <- startMetricsServer (defaultConfig {port = 0}) m []
        threadDelay 200_000
        result <-
          fetchRemoteSubscriptionStatusRows
            (RemoteEndpoint (T.pack ("http://127.0.0.1:" <> show srv.serverPort)))
        stopMetricsServer srv
        case result of
          Left err -> err `shouldSatisfy` T.isInfixOf "404"
          Right rows -> expectationFailure ("expected a 404 error, got rows: " <> show rows)

  it "returns the documented 404 JSON body for unknown paths when mounted standalone" $
    Warp.testWithApplication (pure (subscriptionsApp (pure []))) $ \port -> do
      mgr <- newManager defaultManagerSettings
      req <- parseRequest ("http://127.0.0.1:" <> show port <> "/definitely/not/a/route")
      resp <- httpLbs req mgr
      statusCode (responseStatus resp) `shouldBe` 404
      Aeson.decode (responseBody resp)
        `shouldBe` Just (Aeson.object ["error" Aeson..= ("Not found" :: Text)])

-- | GET a URL and decode the body as @[SubscriptionStatusRow]@ (via the shared codec).
getRows :: String -> IO [SubscriptionStatusRow]
getRows url = do
  mgr <- newManager defaultManagerSettings
  req <- parseRequest url
  resp <- httpLbs req mgr
  case Aeson.eitherDecode (responseBody resp) of
    Right rows -> pure rows
    Left err -> expectationFailure ("could not decode " <> url <> ": " <> err) >> pure []

mkView :: Text -> Int32 -> Text -> Int64 -> SubscriptionStateView
mkView name member phase position =
  SubscriptionStateView
    { subscriptionName = SubscriptionName name,
      member = member,
      state = Live (GlobalPosition position),
      statePhase = phase,
      cursor = GlobalPosition position
    }

ev :: Text -> EventData
ev typ =
  EventData
    { eventId = Nothing,
      eventType = EventType typ,
      payload = Aeson.Null,
      metadata = Nothing,
      causationId = Nothing,
      correlationId = Nothing
    }

waitUntilPhase :: Int -> KirokuStore -> (SubscriptionName, Int32) -> Text -> IO Bool
waitUntilPhase budget store key phase
  | budget <= 0 = pure False
  | otherwise = do
      snapshot <- subscriptionStates store
      case Map.lookup key snapshot of
        Just status | statePhase status == phase -> pure True
        _ -> do
          threadDelay 20_000
          waitUntilPhase (budget - 20_000) store key phase