packages feed

bloodhound-1.0.0.0: tests/Test/MigrationReindexSpec.hs

{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}

module Test.MigrationReindexSpec (spec) where

import Data.Aeson
import Data.ByteString.Lazy.Char8 qualified as LBS
import Database.Bloodhound.Client.Cluster (BackendType (..))
import Database.Bloodhound.ElasticSearch8.Client qualified as ClientES8
import Database.Bloodhound.ElasticSearch8.Requests qualified as RequestsES8
import Database.Bloodhound.ElasticSearch8.Types qualified as Types
import Database.Bloodhound.ElasticSearch9.Requests qualified as RequestsES9
import System.Environment (lookupEnv)
import TestsUtils.Common
import TestsUtils.Import
import Prelude

------------------------------------------------------------------------------
-- Sample payloads (built from the ES9 OpenAPI spec shapes).
------------------------------------------------------------------------------

-- | Canonical @POST /_migration/reindex@ request body: reindex the
-- @my-data-stream@ data stream in @upgrade@ mode.
sampleMigrateReindexRequestBytes :: LBS.ByteString
sampleMigrateReindexRequestBytes =
  "{\
  \  \"source\": {\"index\": \"my-data-stream\"},\
  \  \"mode\": \"upgrade\"\
  \}"

-- | A @GET /_migration/reindex/{index}/_status@ response mid-flight: one
-- index in progress (half its docs reindexed), one errored, the rest
-- pending. @start_time@ is an ISO-8601 string (one of the two wire forms
-- of @_types.DateTime@).
sampleMigrateReindexStatusBytes :: LBS.ByteString
sampleMigrateReindexStatusBytes =
  "{\
  \  \"start_time\": \"2024-01-15T10:30:00.000Z\",\
  \  \"start_time_millis\": 1705314600000,\
  \  \"complete\": false,\
  \  \"total_indices_in_data_stream\": 5,\
  \  \"total_indices_requiring_upgrade\": 4,\
  \  \"successes\": 2,\
  \  \"in_progress\": [\
  \    {\
  \      \"index\": \"logs-app-000003\",\
  \      \"total_doc_count\": 1000,\
  \      \"reindexed_doc_count\": 500\
  \    }\
  \  ],\
  \  \"pending\": 1,\
  \  \"errors\": [\
  \    {\
  \      \"index\": \"logs-app-000001\",\
  \      \"message\": \"cluster:blocked\"\
  \    }\
  \  ]\
  \}"

-- | A completed status with @start_time@ as an epoch-millis number (the
-- other @_types.DateTime@ wire form) and an @exception@ set.
sampleMigrateReindexStatusCompleteBytes :: LBS.ByteString
sampleMigrateReindexStatusCompleteBytes =
  "{\
  \  \"start_time_millis\": 1705314600000,\
  \  \"complete\": true,\
  \  \"total_indices_in_data_stream\": 3,\
  \  \"total_indices_requiring_upgrade\": 2,\
  \  \"successes\": 1,\
  \  \"in_progress\": [],\
  \  \"pending\": 0,\
  \  \"errors\": [],\
  \  \"exception\": \"partial failure\"\
  \}"

------------------------------------------------------------------------------
-- Spec
------------------------------------------------------------------------------

spec :: Spec
spec = describe "Migration reindex API" $ do
  --------------------------------------------------------------------------
  -- MigrateReindexMode
  --------------------------------------------------------------------------
  describe "MigrateReindexMode JSON" $ do
    it "encodes the upgrade constructor" $
      encode Types.MigrateReindexModeUpgrade `shouldBe` "\"upgrade\""

    it "decodes the upgrade wire string" $
      (decode "\"upgrade\"" :: Maybe Types.MigrateReindexMode)
        `shouldBe` Just Types.MigrateReindexModeUpgrade

    it "rejects an unknown mode" $
      (decode "\"downgrade\"" :: Maybe Types.MigrateReindexMode)
        `shouldBe` Nothing

    it "rejects a non-string value" $
      (decode "42" :: Maybe Types.MigrateReindexMode)
        `shouldBe` Nothing

  --------------------------------------------------------------------------
  -- MigrateReindexSource / MigrateReindexRequest
  --------------------------------------------------------------------------
  describe "MigrateReindexRequest JSON" $ do
    it "decodes the canonical ES request body" $
      case (decode sampleMigrateReindexRequestBytes :: Maybe Types.MigrateReindexRequest) of
        Nothing -> expectationFailure "expected MigrateReindexRequest decode"
        Just decoded -> do
          Types.migrateReindexRequestMode decoded
            `shouldBe` Types.MigrateReindexModeUpgrade
          Types.migrateReindexSourceIndex
            (Types.migrateReindexRequestSource decoded)
            `shouldBe` [qqIndexName|my-data-stream|]

    it "round-trips through ToJSON/FromJSON" $ do
      let req =
            Types.mkMigrateReindexRequest [qqIndexName|logs-stream|]
      (decode . encode) req `shouldBe` Just req

    it "mkMigrateReindexRequest sets the upgrade mode" $ do
      let req = Types.mkMigrateReindexRequest [qqIndexName|ds|]
      Types.migrateReindexRequestMode req
        `shouldBe` Types.MigrateReindexModeUpgrade

    it "rejects a body missing the mode field" $
      ( decode
          "{\"source\":{\"index\":\"ds\"}}" ::
          Maybe Types.MigrateReindexRequest
      )
        `shouldBe` Nothing

    it "rejects a body missing the source field" $
      ( decode "{\"mode\":\"upgrade\"}" ::
          Maybe Types.MigrateReindexRequest
      )
        `shouldBe` Nothing

    it "rejects a non-object body" $
      (decode "[]" :: Maybe Types.MigrateReindexRequest)
        `shouldBe` Nothing

  --------------------------------------------------------------------------
  -- MigrateReindexInProgress
  --------------------------------------------------------------------------
  describe "MigrateReindexInProgress JSON" $ do
    it "decodes a canonical entry" $
      case ( decode
               "{\"index\":\"logs-app-000003\",\"total_doc_count\":1000,\"reindexed_doc_count\":500}" ::
               Maybe Types.MigrateReindexInProgress
           ) of
        Nothing -> expectationFailure "expected MigrateReindexInProgress decode"
        Just decoded -> do
          Types.migrateReindexInProgressIndex decoded
            `shouldBe` [qqIndexName|logs-app-000003|]
          Types.migrateReindexInProgressTotalDocCount decoded
            `shouldBe` 1000
          Types.migrateReindexInProgressReindexedDocCount decoded
            `shouldBe` 500

    it "round-trips through ToJSON/FromJSON" $ do
      let entry =
            Types.MigrateReindexInProgress
              { Types.migrateReindexInProgressIndex =
                  [qqIndexName|idx|],
                Types.migrateReindexInProgressTotalDocCount = 7,
                Types.migrateReindexInProgressReindexedDocCount = 3
              }
      (decode . encode) entry `shouldBe` Just entry

    it "rejects a body missing total_doc_count" $
      ( decode
          "{\"index\":\"idx\",\"reindexed_doc_count\":3}" ::
          Maybe Types.MigrateReindexInProgress
      )
        `shouldBe` Nothing

  --------------------------------------------------------------------------
  -- MigrateReindexError
  --------------------------------------------------------------------------
  describe "MigrateReindexError JSON" $ do
    it "decodes a canonical entry" $
      case ( decode
               "{\"index\":\"logs-app-000001\",\"message\":\"cluster:blocked\"}" ::
               Maybe Types.MigrateReindexError
           ) of
        Nothing -> expectationFailure "expected MigrateReindexError decode"
        Just decoded -> do
          Types.migrateReindexErrorIndex decoded
            `shouldBe` [qqIndexName|logs-app-000001|]
          Types.migrateReindexErrorMessage decoded
            `shouldBe` "cluster:blocked"

    it "round-trips through ToJSON/FromJSON" $ do
      let err =
            Types.MigrateReindexError
              { Types.migrateReindexErrorIndex = [qqIndexName|idx|],
                Types.migrateReindexErrorMessage = "boom"
              }
      (decode . encode) err `shouldBe` Just err

  --------------------------------------------------------------------------
  -- MigrateReindexStatus
  --------------------------------------------------------------------------
  describe "MigrateReindexStatus JSON" $ do
    it "decodes the canonical mid-flight response" $
      case (decode sampleMigrateReindexStatusBytes :: Maybe Types.MigrateReindexStatus) of
        Nothing -> expectationFailure "expected MigrateReindexStatus decode"
        Just decoded -> do
          Types.migrateReindexStatusStartTimeMillis decoded
            `shouldBe` 1705314600000
          Types.migrateReindexStatusComplete decoded `shouldBe` False
          Types.migrateReindexStatusTotalIndicesInDataStream decoded
            `shouldBe` 5
          Types.migrateReindexStatusSuccesses decoded `shouldBe` 2
          Types.migrateReindexStatusPending decoded `shouldBe` 1
          length (Types.migrateReindexStatusInProgress decoded)
            `shouldBe` 1
          length (Types.migrateReindexStatusErrors decoded)
            `shouldBe` 1
          -- start_time accepted as an ISO-8601 string
          Types.migrateReindexStatusStartTime decoded
            `shouldSatisfy` isJust
          Types.migrateReindexStatusException decoded
            `shouldBe` Nothing

    it "decodes a completed response with epoch-number start_time omitted and exception set" $
      case (decode sampleMigrateReindexStatusCompleteBytes :: Maybe Types.MigrateReindexStatus) of
        Nothing -> expectationFailure "expected MigrateReindexStatus decode"
        Just decoded -> do
          Types.migrateReindexStatusComplete decoded `shouldBe` True
          Types.migrateReindexStatusStartTime decoded
            `shouldBe` Nothing
          Types.migrateReindexStatusInProgress decoded `shouldBe` []
          Types.migrateReindexStatusErrors decoded `shouldBe` []
          Types.migrateReindexStatusException decoded
            `shouldBe` Just "partial failure"

    it "tolerates missing in_progress and errors (default to empty)" $
      case ( decode
               "{\"start_time_millis\":1,\"complete\":true,\
               \ \"total_indices_in_data_stream\":0,\
               \ \"total_indices_requiring_upgrade\":0,\
               \ \"successes\":0,\"pending\":0}" ::
               Maybe Types.MigrateReindexStatus
           ) of
        Nothing -> expectationFailure "expected MigrateReindexStatus decode"
        Just decoded -> do
          Types.migrateReindexStatusInProgress decoded `shouldBe` []
          Types.migrateReindexStatusErrors decoded `shouldBe` []

    it "round-trips through ToJSON/FromJSON (omitNulls drops empty arrays)" $ do
      -- 'omitNulls' drops empty arrays and Nothing fields, so after a
      -- round-trip an empty @in_progress@/@errors@ and absent @start_time@
      -- resurface as @[]@ / @Nothing@ (the decoder defaults them).
      let s =
            Types.MigrateReindexStatus
              { Types.migrateReindexStatusStartTime = Nothing,
                Types.migrateReindexStatusStartTimeMillis = 1,
                Types.migrateReindexStatusComplete = True,
                Types.migrateReindexStatusTotalIndicesInDataStream = 0,
                Types.migrateReindexStatusTotalIndicesRequiringUpgrade = 0,
                Types.migrateReindexStatusSuccesses = 0,
                Types.migrateReindexStatusInProgress = [],
                Types.migrateReindexStatusPending = 0,
                Types.migrateReindexStatusErrors = [],
                Types.migrateReindexStatusException = Nothing
              }
      (decode . encode) s `shouldBe` Just s

    it "round-trips a populated status unchanged" $
      case (decode sampleMigrateReindexStatusBytes :: Maybe Types.MigrateReindexStatus) of
        Nothing -> expectationFailure "expected MigrateReindexStatus decode"
        Just decoded -> (decode . encode) decoded `shouldBe` Just decoded

    it "rejects a body missing start_time_millis" $
      ( decode
          "{\"complete\":true,\"total_indices_in_data_stream\":0,\
          \ \"total_indices_requiring_upgrade\":0,\"successes\":0,\
          \ \"pending\":0}" ::
          Maybe Types.MigrateReindexStatus
      )
        `shouldBe` Nothing

    it "rejects a non-object body" $
      (decode "42" :: Maybe Types.MigrateReindexStatus)
        `shouldBe` Nothing

  --------------------------------------------------------------------------
  -- CreateIndexFromBody
  --------------------------------------------------------------------------
  describe "CreateIndexFromBody JSON" $ do
    it "decodes a fully-populated body" $
      case ( decode
               "{\"mappings_override\":{\"enabled\":true},\
               \ \"settings_override\":{\"index\":{\"number_of_shards\":1}},\
               \ \"remove_index_blocks\":false}" ::
               Maybe Types.CreateIndexFromBody
           ) of
        Nothing -> expectationFailure "expected CreateIndexFromBody decode"
        Just decoded -> do
          Types.createIndexFromBodyRemoveIndexBlocks decoded
            `shouldBe` Just False
          Types.createIndexFromBodyMappingsOverride decoded
            `shouldSatisfy` isJust

    it "decodes an empty object as the default (all Nothing)" $
      (decode "{}" :: Maybe Types.CreateIndexFromBody)
        `shouldBe` Just Types.defaultCreateIndexFromBody

    it "omits all Nothing fields from ToJSON output" $
      encode Types.defaultCreateIndexFromBody `shouldBe` "{}"

    it "round-trips a fully-populated body" $ do
      let body =
            Types.CreateIndexFromBody
              { Types.createIndexFromBodyMappingsOverride =
                  Just (object ["enabled" .= True]),
                Types.createIndexFromBodySettingsOverride = Nothing,
                Types.createIndexFromBodyRemoveIndexBlocks = Just True
              }
      (decode . encode) body `shouldBe` Just body

  --------------------------------------------------------------------------
  -- CreateIndexFromResponse
  --------------------------------------------------------------------------
  describe "CreateIndexFromResponse JSON" $ do
    it "decodes a canonical response" $ do
      let Just decoded =
            decode
              "{\"acknowledged\":true,\"index\":\"my-new-index\",\
              \ \"shards_acknowledged\":true}" ::
              Maybe Types.CreateIndexFromResponse
      Types.createIndexFromResponseAcknowledged decoded `shouldBe` True
      Types.createIndexFromResponseIndex decoded
        `shouldBe` [qqIndexName|my-new-index|]
      Types.createIndexFromResponseShardsAcknowledged decoded
        `shouldBe` True

    it "round-trips through ToJSON/FromJSON" $ do
      let resp =
            Types.CreateIndexFromResponse
              { Types.createIndexFromResponseAcknowledged = True,
                Types.createIndexFromResponseIndex = [qqIndexName|dest|],
                Types.createIndexFromResponseShardsAcknowledged = False
              }
      (decode . encode) resp `shouldBe` Just resp

    it "rejects a body missing shards_acknowledged" $
      ( decode
          "{\"acknowledged\":true,\"index\":\"dest\"}" ::
          Maybe Types.CreateIndexFromResponse
      )
        `shouldBe` Nothing

  --------------------------------------------------------------------------
  -- Endpoint shape tests (ES8 builders — the canonical implementation)
  --------------------------------------------------------------------------
  describe "migrateReindex endpoint shape" $ do
    let req = Types.mkMigrateReindexRequest [qqIndexName|my-data-stream|]
        bhReq = RequestsES8.migrateReindex req

    it "POSTs /_migration/reindex" $
      getRawEndpoint (bhRequestEndpoint bhReq)
        `shouldBe` ["_migration", "reindex"]

    it "uses the POST method" $
      bhRequestMethod bhReq `shouldBe` "POST"

    it "carries the encoded request body" $ do
      bhRequestBody bhReq `shouldSatisfy` isJust
      -- the body round-trips back into the same request
      let Just body = bhRequestBody bhReq
      (decode body :: Maybe Types.MigrateReindexRequest) `shouldBe` Just req

    it "carries no query string" $
      getRawEndpointQueries (bhRequestEndpoint bhReq) `shouldBe` []

  describe "cancelMigrateReindex endpoint shape" $ do
    let bhReq =
          RequestsES8.cancelMigrateReindex
            ([qqIndexName|my-data-stream|] :| [])

    it "POSTs /_migration/reindex/{index}/_cancel" $
      getRawEndpoint (bhRequestEndpoint bhReq)
        `shouldBe` ["_migration", "reindex", "my-data-stream", "_cancel"]

    it "uses the POST method" $
      bhRequestMethod bhReq `shouldBe` "POST"

    it "sends an empty body" $
      bhRequestBody bhReq `shouldBe` Just ""

    it "comma-joins multiple indices into one path segment" $ do
      let bhReq' =
            RequestsES8.cancelMigrateReindex
              ( [qqIndexName|ds-one|]
                  :| [[qqIndexName|ds-two|]]
              )
      getRawEndpoint (bhRequestEndpoint bhReq')
        `shouldBe` ["_migration", "reindex", "ds-one,ds-two", "_cancel"]

  describe "getMigrateReindexStatus endpoint shape" $ do
    let bhReq =
          RequestsES8.getMigrateReindexStatus
            ([qqIndexName|my-data-stream|] :| [])

    it "GETs /_migration/reindex/{index}/_status" $
      getRawEndpoint (bhRequestEndpoint bhReq)
        `shouldBe` ["_migration", "reindex", "my-data-stream", "_status"]

    it "uses the GET method" $
      bhRequestMethod bhReq `shouldBe` "GET"

    it "carries no request body" $
      bhRequestBody bhReq `shouldBe` Nothing

  describe "createIndexFrom endpoint shape" $ do
    let bhReq = RequestsES8.createIndexFrom [qqIndexName|src|] [qqIndexName|dst|]

    it "POSTs /_create_from/{source}/{dest}" $
      getRawEndpoint (bhRequestEndpoint bhReq)
        `shouldBe` ["_create_from", "src", "dst"]

    it "uses the POST method" $
      bhRequestMethod bhReq `shouldBe` "POST"

    it "sends an empty body when no overrides are supplied" $
      bhRequestBody bhReq `shouldBe` Just ""

  describe "createIndexFromWith endpoint shape" $ do
    let body =
          Types.CreateIndexFromBody
            { Types.createIndexFromBodyMappingsOverride =
                Just (object ["enabled" .= True]),
              Types.createIndexFromBodySettingsOverride = Nothing,
              Types.createIndexFromBodyRemoveIndexBlocks = Just False
            }
        bhReq =
          RequestsES8.createIndexFromWith
            [qqIndexName|src|]
            [qqIndexName|dst|]
            (Just body)

    it "POSTs /_create_from/{source}/{dest}" $
      getRawEndpoint (bhRequestEndpoint bhReq)
        `shouldBe` ["_create_from", "src", "dst"]

    it "carries the encoded override body" $ do
      bhRequestBody bhReq `shouldSatisfy` isJust
      let Just b = bhRequestBody bhReq
      (decode b :: Maybe Types.CreateIndexFromBody) `shouldBe` Just body

  describe "ES9 aliases match the ES8 builders" $ do
    let indices = [qqIndexName|my-data-stream|] :| []

    it "migrateReindex produces the same endpoint as ES8" $ do
      let req = Types.mkMigrateReindexRequest [qqIndexName|ds|]
      getRawEndpoint
        (bhRequestEndpoint (RequestsES9.migrateReindex req))
        `shouldBe` getRawEndpoint
          (bhRequestEndpoint (RequestsES8.migrateReindex req))

    it "getMigrateReindexStatus produces the same endpoint as ES8" $
      getRawEndpoint
        (bhRequestEndpoint (RequestsES9.getMigrateReindexStatus indices))
        `shouldBe` getRawEndpoint
          (bhRequestEndpoint (RequestsES8.getMigrateReindexStatus indices))

    it "cancelMigrateReindex produces the same endpoint as ES8" $
      getRawEndpoint
        (bhRequestEndpoint (RequestsES9.cancelMigrateReindex indices))
        `shouldBe` getRawEndpoint
          (bhRequestEndpoint (RequestsES8.cancelMigrateReindex indices))

    it "createIndexFrom produces the same endpoint as ES8" $
      getRawEndpoint
        ( bhRequestEndpoint
            (RequestsES9.createIndexFrom [qqIndexName|s|] [qqIndexName|d|])
        )
        `shouldBe` getRawEndpoint
          ( bhRequestEndpoint
              (RequestsES8.createIndexFrom [qqIndexName|s|] [qqIndexName|d|])
          )

  --------------------------------------------------------------------------
  -- Live integration (gated on ES8/ES9; mutating ops further gated
  -- behind ES_TEST_RUN_MUTATING so cluster state is not perturbed by
  -- default).
  --------------------------------------------------------------------------
  backendSpecific
    [ElasticSearch8, ElasticSearch9]
    $ describe "Migration reindex API (live integration)"
    $ do
      it "getMigrateReindexStatus returns a parseable response when ES_TEST_RUN_MUTATING=true" $
        withTestEnv $ do
          -- Querying a non-existent data stream surfaces as an EsError
          -- (404) rather than a decodable status, so this is gated on
          -- the mutating flag alongside the cluster-perturbing ops; a
          -- meaningful live assertion requires a real data stream.
          enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
          case enabled of
            Just "true" -> do
              result <-
                ClientES8.getMigrateReindexStatus
                  ([qqIndexName|bloodhound-missing-stream|] :| [])
              liftIO $
                Types.migrateReindexStatusComplete result
                  `shouldSatisfy` (const True :: Bool -> Bool)
            _ ->
              liftIO $
                pendingWith
                  "skipped: set ES_TEST_RUN_MUTATING=true to run this test"

      it "migrateReindex returns Acknowledged when ES_TEST_RUN_MUTATING=true" $
        withTestEnv $ do
          enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
          case enabled of
            Just "true" -> do
              ack <-
                ClientES8.migrateReindex
                  ( Types.mkMigrateReindexRequest
                      [qqIndexName|bloodhound-missing-stream|]
                  )
              liftIO $ isAcknowledged ack `shouldBe` True
            _ ->
              liftIO $
                pendingWith
                  "skipped: set ES_TEST_RUN_MUTATING=true to run this mutating test"

      it "createIndexFrom returns acknowledged when ES_TEST_RUN_MUTATING=true" $
        withTestEnv $ do
          enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
          case enabled of
            Just "true" -> do
              resp <-
                ClientES8.createIndexFrom
                  [qqIndexName|bloodhound-tests-twitter-1|]
                  [qqIndexName|bloodhound-createfrom-dest|]
              liftIO $
                Types.createIndexFromResponseAcknowledged resp
                  `shouldBe` True
            _ ->
              liftIO $
                pendingWith
                  "skipped: set ES_TEST_RUN_MUTATING=true to run this mutating test"