{-# OPTIONS_GHC -Wno-ambiguous-fields #-}

-- This file is part of the Wire Server implementation.
--
-- Copyright (C) 2025 Wire Swiss GmbH <opensource@wire.com>
--
-- This program is free software: you can redistribute it and/or modify it under
-- the terms of the GNU Affero General Public License as published by the Free
-- Software Foundation, either version 3 of the License, or (at your option) any
-- later version.
--
-- This program is distributed in the hope that it will be useful, but WITHOUT
-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
-- details.
--
-- You should have received a copy of the GNU Affero General Public License along
-- with this program. If not, see <https://www.gnu.org/licenses/>.

module Test.Events where

import API.Brig
import API.BrigCommon
import API.Common
import API.Galley
import API.Gundeck
import qualified API.GundeckInternal as GundeckInternal
import qualified Control.Concurrent.Timeout as Timeout
import Control.Lens ((.~), (^?!))
import Control.Monad.Codensity
import Control.Monad.RWS (asks)
import Control.Monad.Trans.Class
import Control.Monad.Trans.Maybe
import Control.Retry
import Data.ByteString.Conversion (toByteString')
import qualified Data.ProtoLens as Proto
import Data.ProtoLens.Labels ()
import Data.Proxy (Proxy (..))
import qualified Data.Text as Text
import Data.Timeout
import Network.AMQP.Extended
import Network.RabbitMqAdmin
import qualified Network.WebSockets as WS
import Notifications
import Numeric.Lens
import qualified Proto.Otr as Proto
import qualified Proto.Otr_Fields as Proto
import Servant.API (AsApi, ToServant, toServant)
import Servant.API.Generic (fromServant)
import Servant.Client (AsClientT)
import qualified Servant.Client as Servant
import SetupHelpers
import Testlib.Prelude
import Testlib.ResourcePool
import UnliftIO hiding (handle)

testConsumeEventsOneWebSocket :: (HasCallStack) => App ()
testConsumeEventsOneWebSocket = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def

  lastNotifResp <-
    retrying
      (constantDelay 10_000 <> limitRetries 10)
      (\_ resp -> pure $ resp.status == 404)
      (\_ -> getLastNotification alice def)
  lastNotifId <- lastNotifResp.json %. "id" & asString

  client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  clientId <- objId client

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    deliveryTag <- assertEvent ws $ \e -> do
      e %. "type" `shouldMatch` "event"
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
      e %. "data.delivery_tag"
    assertNoEvent_ ws

    sendAck ws deliveryTag False
    assertNoEvent_ ws

    handle <- randomHandle
    putHandle alice handle >>= assertSuccess

    assertEvent ws $ \e -> do
      e %. "type" `shouldMatch` "event"
      e %. "data.event.payload.0.type" `shouldMatch` "user.update"
      e %. "data.event.payload.0.user.handle" `shouldMatch` handle

  -- No new notifications should be stored in Cassandra as the user doesn't have
  -- any legacy clients
  getNotifications alice def {since = Just lastNotifId} `bindResponse` \resp -> do
    resp.status `shouldMatchInt` 200
    shouldBeEmpty $ resp.json %. "notifications"

testWebSocketTimeout :: (HasCallStack) => App ()
testWebSocketTimeout = withModifiedBackend
  ( enableConsumableNotifications
      def
        { cannonCfg =
            setField "wsOpts.activityTimeout" (1000000 :: Int)
              >=> setField "wsOpts.pongTimeout" (1000000 :: Int)
        }
  )
  $ \domain -> do
    alice <- randomUser domain def
    client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    clientId <- objId client

    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
      deliveryTag <- assertEvent ws $ \e -> do
        e %. "type" `shouldMatch` "event"
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "data.event.payload.0.client.id" `shouldMatch` clientId
        e %. "data.delivery_tag"
      sendAck ws deliveryTag False

      result <- timeout 2500000 (killWebSocketClient ws)
      when (isNothing result)
        $ assertFailure "Expected web socket timeout"

testConsumeTempEvents :: (HasCallStack) => App ()
testConsumeTempEvents = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def

  client0 <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  clientId0 <- objId client0

  lowerCodensity $ do
    ws0 <- createEventsWebSocket alice (Just clientId0)

    -- Ensure there is no race between event for this client being pushed and temp
    -- consumer being created
    lift $ do
      expectAndAckNewClientEvent ws0 clientId0
      assertNoEvent_ ws0

    wsTemp <- createEventsWebSocket alice Nothing

    lift $ do
      client1 <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
      clientId1 <- objId client1

      -- Temp client gets this event as it happens after temp client has started
      -- listening
      void $ expectAndAckNewClientEvent wsTemp clientId1

      -- Client0 should also be notified even if there is a temp client
      void $ expectAndAckNewClientEvent ws0 clientId1

      assertNoEvent_ wsTemp
      assertNoEvent_ ws0
  where
    expectAndAckNewClientEvent :: EventWebSocket -> String -> App ()
    expectAndAckNewClientEvent ws cid =
      assertEvent ws $ \e -> do
        e %. "type" `shouldMatch` "event"
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "data.event.payload.0.client.id" `shouldMatch` cid

        ackEvent ws e

testTemporaryQueuesAreDeletedAfterUse :: (HasCallStack) => App ()
testTemporaryQueuesAreDeletedAfterUse = do
  startDynamicBackendsReturnResources [enableConsumableNotifications def] $ \[beResource] -> do
    let domain = beResource.berDomain
    rabbitmqAdmin <- mkRabbitMqAdminClientForResource beResource
    [alice, bob] <- createAndConnectUsers [domain, domain]

    -- Create client for alice, so their temp websocket works
    aliceClient <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    aliceId <- asString $ alice %. "qualified_id.id"
    aliceClientId <- asString $ aliceClient %. "id"

    let aliceClientQueueName = "user-notifications." <> aliceId <> "." <> aliceClientId
        aliceClientQueue = Queue {name = fromString aliceClientQueueName, vhost = fromString beResource.berVHost}
        deadNotifsQueue = Queue {name = fromString "dead-user-notifications", vhost = fromString beResource.berVHost}
        cellsEventsQueue = Queue {name = fromString "cells_events", vhost = fromString beResource.berVHost}
        backgroundJobsQueue = Queue {name = fromString "background-jobs", vhost = fromString beResource.berVHost}

    -- Wait for queue for the new client to be created
    eventually $ do
      queuesBeforeWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
      queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue]

    runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do
      handle <- randomHandle
      putHandle bob handle >>= assertSuccess

      queuesDuringWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
      addJSONToFailureContext "queuesDuringWS" queuesDuringWS $ do
        length queuesDuringWS.items `shouldMatchInt` 5

      -- We cannot use 'assertEvent' here because there is a race between the temp
      -- queue being created and rabbitmq fanning out the previous events.
      void $ assertFindsEvent ws $ \e -> do
        e %. "type" `shouldMatch` "event"
        e %. "data.event.payload.0.type" `shouldMatch` "user.update"
        e %. "data.event.payload.0.user.id" `shouldMatch` objId bob
        e %. "data.event.payload.0.user.handle" `shouldMatch` handle

        ackEvent ws e

    eventually $ do
      queuesAfterWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1
      queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue]

testSendMessageNoReturnToSenderWithConsumableNotificationsProteus :: (HasCallStack) => App ()
testSendMessageNoReturnToSenderWithConsumableNotificationsProteus = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, tid, bob : _) <- createTeam domain 2
  aliceOldClient <- addClient alice def >>= getJSON 201
  aliceClient <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  aliceClientId <- objId aliceClient
  bobClient <- addClient bob def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  bobClientId <- objId bobClient
  conv <- postConversation alice defProteus {team = Just tid, qualifiedUsers = [bob]} >>= getJSON 201
  msg <- mkProteusRecipients alice [(bob, [bobClient]), (alice, [aliceOldClient])] "hello, bob"

  let protoMsg =
        Proto.defMessage @Proto.QualifiedNewOtrMessage
          & #sender . Proto.client .~ (aliceClientId ^?! hex)
          & #recipients .~ [msg]
          & #reportAll .~ Proto.defMessage
  postProteusMessage alice conv protoMsg >>= assertSuccess

  runCodensity (createEventsWebSocket bob (Just bobClientId)) $ \ws -> do
    assertFindsEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "conversation.otr-message-add"
      e %. "data.event.payload.0.data.text" `shouldMatchBase64` fromString "hello, bob"
      ackEvent ws e

  runCodensity (createEventsWebSocket alice (Just aliceClientId)) $ \ws -> do
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      ackEvent ws e
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "conversation.create"
      ackEvent ws e
    assertNoEvent_ ws

testEventsForSpecificClients :: (HasCallStack) => App ()
testEventsForSpecificClients = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  uid <- objId alice
  client1 <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  cid1 <- objId client1
  client2 <- addClient alice def >>= getJSON 201
  cid2 <- objId client2

  lowerCodensity $ do
    ws1 <- createEventsWebSocket alice (Just cid1)
    wsTemp <- createEventsWebSocket alice Nothing
    lift $ do
      void $ consumeAllEvents ws1

      let eventForClient1 =
            object
              [ "recipients" .= [object ["user_id" .= uid, "clients" .= [cid1], "route" .= "any"]],
                "payload" .= [object ["hello" .= "client1"]]
              ]
          eventForClient2 =
            object
              [ "recipients" .= [object ["user_id" .= uid, "clients" .= [cid2], "route" .= "any"]],
                "payload" .= [object ["hello" .= "client2"]]
              ]

      GundeckInternal.postPush domain [eventForClient1, eventForClient2] >>= assertSuccess
      assertEvent ws1 $ \e ->
        e %. "data.event.payload.0.hello" `shouldMatch` "client1"

      addFailureContext "client 1 should not get any events meant for client 2"
        $ assertNoEvent_ ws1

      addFailureContext "temp client should not get any events meant solely for client 1 or 2"
        $ assertNoEvent_ wsTemp

testConsumeEventsForDifferentUsers :: (HasCallStack) => App ()
testConsumeEventsForDifferentUsers = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  bob <- randomUser domain def

  aliceClient <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  aliceClientId <- objId aliceClient

  bobClient <- addClient bob def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  bobClientId <- objId bobClient

  lowerCodensity $ do
    aliceWS <- createEventsWebSocket alice (Just aliceClientId)
    bobWS <- createEventsWebSocket bob (Just bobClientId)
    lift $ assertClientAdd aliceClientId aliceWS
    lift $ assertClientAdd bobClientId bobWS
  where
    assertClientAdd :: (HasCallStack) => String -> EventWebSocket -> App ()
    assertClientAdd clientId ws = do
      deliveryTag <- assertEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "data.event.payload.0.client.id" `shouldMatch` clientId
        e %. "data.delivery_tag"
      assertNoEvent_ ws
      sendAck ws deliveryTag False

testConsumeEventsWhileHavingLegacyClients :: (HasCallStack) => App ()
testConsumeEventsWhileHavingLegacyClients = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def

  -- Even if alice has no clients, the notifications should still be persisted
  -- in Cassandra. This choice is kinda arbitrary as these notifications
  -- probably don't mean much, however, it ensures backwards compatibility.
  lastNotifId <-
    awaitNotification alice noValue (const $ pure True) >>= \notif -> do
      notif %. "payload.0.type" `shouldMatch` "user.activate"
      -- There is only one notification (at the time of writing), so we assume
      -- it to be the last one.
      notif %. "id" & asString

  oldClient <- addClient alice def {acapabilities = Just []} >>= getJSON 201

  withWebSocket (alice, "anything-but-conn", oldClient %. "id") $ \oldWS -> do
    newClient <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    newClientId <- newClient %. "id" & asString

    oldNotif <- awaitMatch isUserClientAddNotif oldWS
    oldNotif %. "payload.0.client.id" `shouldMatch` newClientId

    runCodensity (createEventsWebSocket alice (Just newClientId)) $ \ws -> do
      assertEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "data.event.payload.0.client.id" `shouldMatch` newClientId

  -- All notifs are also in Cassandra because of the legacy client
  getNotifications alice def {since = Just lastNotifId} `bindResponse` \resp -> do
    resp.status `shouldMatchInt` 200
    resp.json %. "notifications.0.payload.0.type" `shouldMatch` "user.client-add"
    resp.json %. "notifications.1.payload.0.type" `shouldMatch` "user.client-add"

testConsumeEventsAcks :: (HasCallStack) => App ()
testConsumeEventsAcks = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  clientId <- objId client

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId

  -- without ack, we receive the same event again
  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    deliveryTag <- assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
      e %. "data.delivery_tag"
    sendAck ws deliveryTag False

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertNoEvent_ ws

testConsumeEventsMultipleAcks :: (HasCallStack) => App ()
testConsumeEventsMultipleAcks = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  clientId <- objId client

  handle <- randomHandle
  putHandle alice handle >>= assertSuccess

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId

    deliveryTag <- assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.update"
      e %. "data.event.payload.0.user.handle" `shouldMatch` handle
      e %. "data.delivery_tag"

    sendAck ws deliveryTag True

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertNoEvent_ ws

testConsumeEventsAckNewEventWithoutAckingOldOne :: (HasCallStack) => App ()
testConsumeEventsAckNewEventWithoutAckingOldOne = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  clientId <- objId client

  handle <- randomHandle
  putHandle alice handle >>= assertSuccess

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId

    deliveryTagHandleAdd <- assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.update"
      e %. "data.event.payload.0.user.handle" `shouldMatch` handle
      e %. "data.delivery_tag"

    -- Only ack the handle add delivery tag
    sendAck ws deliveryTagHandleAdd False

  -- Expect client-add event to be delivered again.
  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    deliveryTagClientAdd <- assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
      e %. "data.delivery_tag"

    sendAck ws deliveryTagClientAdd False

  runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
    assertNoEvent_ ws

testEventsDeadLettered :: (HasCallStack) => App ()
testEventsDeadLettered = do
  let notifTTL = 1 # Second
  withModifiedBackend (enableConsumableNotifications (def {gundeckCfg = setField "settings.notificationTTL" (notifTTL #> Second)})) $ \domain -> do
    alice <- randomUser domain def

    -- This generates an event
    client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    clientId <- objId client

    -- We expire the add client event by waiting it out
    Timeout.threadDelay (notifTTL + 500 # MilliSecond)

    -- Generate a second event
    handle1 <- randomHandle
    putHandle alice handle1 >>= assertSuccess

    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
      assertEvent ws $ \e -> do
        e %. "type" `shouldMatch` "notifications_missed"

      -- Until we ack the full sync, we can't get new events
      ackFullSync ws

      -- withEventsWebSocket alice clientId $ \eventsChan ackChan -> do
      -- Now we can see the next event
      assertEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.update"
        e %. "data.event.payload.0.user.handle" `shouldMatch` handle1
        ackEvent ws e

      -- We've consumed the whole queue.
      assertNoEvent_ ws

testEventsDeadLetteredWithReconnect :: (HasCallStack) => App ()
testEventsDeadLetteredWithReconnect = do
  let notifTTL = 1 # Second
  startDynamicBackendsReturnResources [enableConsumableNotifications (def {gundeckCfg = setField "settings.notificationTTL" (notifTTL #> Second)})] $ \[resources] -> do
    let domain :: String = resources.berDomain
    alice <- randomUser domain def

    -- This generates an event
    client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    clientId <- objId client

    -- Force a reconnect by deleting the existing connection
    killAllDeadUserNotificationRabbitMqConns resources

    -- We expire the add client event by waiting it out
    Timeout.threadDelay (notifTTL + 500 # MilliSecond)

    -- Generate a second event
    handle1 <- randomHandle
    putHandle alice handle1 >>= assertSuccess

    runCodensity (createEventsWebSocketWithSync alice (Just clientId)) $ \(endMarker, ws) -> do
      assertEvent ws $ \e -> do
        e %. "type" `shouldMatch` "notifications_missed"

      -- Until we ack the full sync, we can't get new events
      ackFullSync ws

      -- Now we can see the next event
      assertEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.update"
        e %. "data.event.payload.0.user.handle" `shouldMatch` handle1
        ackEvent ws e

      -- We've consumed the whole queue.
      assertEndOfIniitalSync ws endMarker
  where
    killAllDeadUserNotificationRabbitMqConns :: (HasCallStack) => BackendResource -> App ()
    killAllDeadUserNotificationRabbitMqConns backend = do
      rabbitmqAdminClient <- mkRabbitMqAdminClientForResource backend
      connections <- eventually $ do
        conns <- getDeadUserNotificationConnections rabbitmqAdminClient backend.berVHost
        assertAtLeastOne conns
        pure conns
      for_ connections $ \connection -> do
        rabbitmqAdminClient.deleteConnection connection.name

    getDeadUserNotificationConnections :: (HasCallStack) => AdminAPI (AsClientT App) -> String -> App [Connection]
    getDeadUserNotificationConnections rabbitmqAdminClient vhost = do
      connections <- rabbitmqAdminClient.listConnectionsByVHost (Text.pack vhost)
      pure $ filter (\c -> Just (fromString "dead-user-notifications-watcher") == c.userProvidedName) connections

testTransientEventsDoNotTriggerDeadLetters :: (HasCallStack) => App ()
testTransientEventsDoNotTriggerDeadLetters = do
  let notifTTL = 1 # Second
  withModifiedBackend (enableConsumableNotifications (def {gundeckCfg = setField "settings.notificationTTL" (notifTTL #> Second)})) $ \domain -> do
    alice <- randomUser domain def
    -- Creates a non-transient event
    client <- addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
    clientId <- objId client

    -- consume it
    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
      assertFindsEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "type" `shouldMatch` "event"
        e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
        e %. "data.event.payload.0.client.id" `shouldMatch` clientId
        deliveryTag <- e %. "data.delivery_tag"
        sendAck ws deliveryTag False

    -- Self conv ID is same as user's ID, we'll use this to send typing
    -- indicators, so we don't have to create another conv.
    selfConvId <- objQidObject alice
    -- Typing status is transient, currently no one is listening.
    sendTypingStatus alice selfConvId "started" >>= assertSuccess

    runCodensity (createEventsWebSocket alice (Just clientId)) $ \ws -> do
      assertNoEvent_ ws

testTransientEvents :: (HasCallStack) => App ()
testTransientEvents = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, _, _) <- mkUserPlusClientWithDomain domain
  (bob, _, bobClient) <- mkUserPlusClientWithDomain domain
  connectTwoUsers alice bob
  bobClientId <- objId bobClient

  conv <- postConversation alice defProteus {qualifiedUsers = [bob]} >>= getJSON 201

  runCodensity (createEventsWebSocketWithSync bob (Just bobClientId)) $ \(marker, bobWs) -> do
    void $ consumeEventsUntilEndOfInitialSync bobWs marker

    sendTypingStatus alice conv "started" >>= assertSuccess

    assertEvent bobWs $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "conversation.typing"
      e %. "data.event.payload.0.qualified_conversation" `shouldMatch` (conv %. "qualified_id")
      ackEvent bobWs e

  handle1 <- randomHandle
  putHandle bob handle1 >>= assertSuccess

  sendTypingStatus alice conv "stopped" >>= assertSuccess

  handle2 <- randomHandle
  putHandle bob handle2 >>= assertSuccess

  -- We shouldn't see the stopped typing status because we were not connected to
  -- the websocket when it was sent. The other events should still show up in
  -- order.
  runCodensity (createEventsWebSocket bob (Just bobClient)) $ \ws -> do
    for_ [handle1, handle2] $ \handle ->
      assertEvent ws $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "user.update"
        e %. "data.event.payload.0.user.handle" `shouldMatch` handle
        ackEvent ws e

    assertNoEvent_ ws

testChannelLimit :: (HasCallStack) => App ()
testChannelLimit = withModifiedBackend
  ( enableConsumableNotifications
      ( def
          { cannonCfg =
              setField "rabbitMqMaxChannels" (2 :: Int)
                >=> setField "rabbitMqMaxConnections" (1 :: Int)
          }
      )
  )
  $ \domain -> do
    alice <- randomUser domain def
    (client0 : clients) <-
      replicateM 3
        $ addClient alice def {acapabilities = Just ["consumable-notifications"]}
        >>= getJSON 201
        >>= (%. "id")
        >>= asString

    lowerCodensity $ do
      for_ clients $ \c -> do
        ws <- createEventsWebSocket alice (Just c)
        lift $ assertEvent ws $ \e -> do
          e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
          e %. "data.event.payload.0.client.id" `shouldMatch` c

      -- the first client fails to connect because the server runs out of channels
      do
        eithWS <- createEventsWebSocketEither alice (Just client0) Nothing
        case eithWS of
          Left (WS.MalformedResponse respHead _) ->
            lift $ respHead.responseCode `shouldMatchInt` 503
          Left e ->
            lift $ assertFailure $ "Expected websocket to fail with response code 503, got some other handshake exception: " <> displayException e
          Right _ -> lift $ assertFailure "Expected websocket hanshake to fail, but it didn't"

testChannelKilled :: (HasCallStack) => App ()
testChannelKilled = do
  pool <- asks (.resourcePool)
  runCodensity (acquireResources 1 pool) $ \[backend] -> do
    -- Some times RabbitMQ still remembers connections from previous uses of the
    -- dynamic backend. So we wait to ensure that we kill connection only for our
    -- current.
    void $ killAllRabbitMqConns backend
    waitUntilNoRabbitMqConns backend

    runCodensity (startDynamicBackend backend (enableConsumableNotifications def)) $ \_ -> do
      let domain = backend.berDomain
      alice <- randomUser domain def
      [c1, c2] <-
        replicateM 2
          $ addClient alice def {acapabilities = Just ["consumable-notifications"]}
          >>= getJSON 201
          >>= (%. "id")
          >>= asString

      runCodensity (createEventsWebSocket alice (Just c1)) $ \ws -> do
        -- If creating the user takes longer (async) than adding the clients, we get a
        -- `"user.activate"` here, so we use `assertFindsEvent`.
        assertFindsEvent ws $ \e -> do
          e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
          e %. "data.event.payload.0.client.id" `shouldMatch` c1
          ackEvent ws e

        assertEvent ws $ \e -> do
          e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
          e %. "data.event.payload.0.client.id" `shouldMatch` c2
          ackEvent ws e

        -- The RabbitMQ admin API takes some time to see new connections, so we need
        -- to try a few times.
        recoverAll (constantDelay 500_000 <> limitRetries 10) $ \_ -> do
          conns <- killAllRabbitMqConns backend
          assertAtLeastOne conns

        waitUntilNoRabbitMqConns backend

        assertNoEventHelper ws `shouldMatch` WebSocketDied

testSingleConsumer :: (HasCallStack) => App ()
testSingleConsumer = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  alice <- randomUser domain def
  clientId <-
    addClient alice def {acapabilities = Just ["consumable-notifications"]}
      >>= getJSON 201
      >>= objId

  -- add a second client in order to generate one more notification
  clientId' <- addClient alice def >>= getJSON 201 >>= objId

  lowerCodensity $ do
    ws <- createEventsWebSocket alice (Just clientId)
    ws' <- createEventsWebSocket alice (Just clientId)

    -- the second websocket should get no notifications as long as the first
    -- one is connected
    lift $ assertNoEvent_ ws'

    deliveryTag1 <- lift $ assertEvent ws $ \e -> do
      e %. "type" `shouldMatch` "event"
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId
      e %. "data.delivery_tag"

    lift $ assertNoEvent_ ws'

    lift $ sendAck ws deliveryTag1 False
    lift $ assertNoEvent_ ws'

    deliveryTag2 <- lift $ assertEvent ws $ \e -> do
      e %. "type" `shouldMatch` "event"
      e %. "data.event.payload.0.type" `shouldMatch` "user.client-add"
      e %. "data.event.payload.0.client.id" `shouldMatch` clientId'
      e %. "data.delivery_tag"
    lift $ sendAck ws deliveryTag2 False

    lift $ assertNoEvent_ ws'

testPrefetchCount :: (HasCallStack) => App ()
testPrefetchCount = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, uid, cid) <- mkUserPlusClientWithDomain domain
  emptyQueue alice cid

  for_ [1 :: Int .. 550] $ \i ->
    do
      let event =
            object
              [ "recipients" .= [object ["user_id" .= uid, "clients" .= [cid], "route" .= "any"]],
                "payload" .= [object ["no" .= show i]]
              ]
      GundeckInternal.postPush domain [event] >>= assertSuccess
  runCodensity (createEventsWebSocketWithSync alice (Just cid)) \(endMarker, ws) -> do
    es <- consumeAllEventsNoAck ws
    assertBool ("First 500 events expected, got " ++ show (length es)) $ length es == 500

    forM_ es (ackEvent ws)

    es' <- consumeEventsUntilEndOfInitialSync ws endMarker
    assertBool "Receive at least one outstanding event" $ not (null es')

testEndOfInitialSync :: (HasCallStack) => App ()
testEndOfInitialSync = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, uid, cid) <- mkUserPlusClientWithDomain domain
  let n = 20
  replicateM_ n $ do
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess

  -- marker0 <- randomId
  runCodensity (createEventsWebSocketWithSync alice (Just cid)) \(endMarker, ws) -> do
    preExistingEvents <- consumeEventsUntilEndOfInitialSync ws endMarker
    otherEvents <- consumeAllEvents ws

    -- we expect one user.client-add event, n more events, and one sync event
    length (preExistingEvents <> otherEvents) `shouldMatchInt` (n + 2)

    -- more events should not be followed by the sync event
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "test"
      ackEvent ws e
    assertNoEvent_ ws

  -- when we reconnect, there are no messages but we still get the
  -- synchronization notification.
  runCodensity (createEventsWebSocketWithSync alice (Just cid)) \(endMarker, ws) -> do
    preExistingEvents <- consumeEventsUntilEndOfInitialSync ws endMarker
    otherEvents <- consumeAllEvents ws
    let events = preExistingEvents <> otherEvents
    length events `shouldMatchInt` 1

    -- more events should not be followed by synchronization event
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess
    assertEvent ws $ \e -> do
      e %. "data.event.payload.0.type" `shouldMatch` "test"
      ackEvent ws e
    assertNoEvent_ ws

testEndOfInitialSyncMoreEventsAfterSyncMessage :: (HasCallStack) => App ()
testEndOfInitialSyncMoreEventsAfterSyncMessage = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, uid, cid) <- mkUserPlusClientWithDomain domain
  let n = 20
  replicateM_ n $ do
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess

  runCodensity (createEventsWebSocketWithSync alice (Just cid)) \(endMarker, ws) -> do
    -- it seems this is needed to reduce flakiness,
    -- when the messages below are pushed faster than the sync message is inserted
    Timeout.threadDelay (1 # Second)

    -- before consuming, we push n more events
    replicateM_ n $ do
      GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess

    preExistingEvents <- consumeEventsUntilEndOfInitialSync ws endMarker
    otherEvents <- consumeAllEvents ws

    length (preExistingEvents <> otherEvents) `shouldMatchInt` (n + n + 2)
    addFailureContext ("We should have received " <> show n <> " more events after the sync event but got " <> show (length otherEvents))
      $ (length otherEvents >= n)
      `shouldMatch` True

testEndOfInitialSyncIgnoreExpired :: (HasCallStack) => App ()
testEndOfInitialSyncIgnoreExpired = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, uid, cid) <- mkUserPlusClientWithDomain domain
  let n = 20
  replicateM_ n $ do
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess

  replicateM_ n $ do
    GundeckInternal.postPush domain [mkEvent uid cid True] >>= assertSuccess

  -- Wait for transient events to expire
  Timeout.threadDelay (1 # Second)

  runCodensity (createEventsWebSocketWithSync alice (Just cid)) $ \(endMarker, ws) -> do
    preExistingEvents <- consumeEventsUntilEndOfInitialSync ws endMarker
    otherEvents <- consumeAllEvents ws
    let events = preExistingEvents <> otherEvents
    length events `shouldMatchInt` (n + 2) -- +1 for the sync event, +1 for the client add event

testEndOfInitialSyncAckMultiple :: (HasCallStack) => App ()
testEndOfInitialSyncAckMultiple = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
  (alice, uid, cid) <- mkUserPlusClientWithDomain domain
  let n = 20
  replicateM_ n $ do
    GundeckInternal.postPush domain [mkEvent uid cid False] >>= assertSuccess

  runCodensity (createEventsWebSocketWithSync alice (Just cid)) $ \(endMarker, ws) -> do
    void $ assertEvent ws pure
    e <- assertEvent ws pure
    dt <- e %. "data.delivery_tag"
    -- we ack the first 2 events with one ack
    sendAck ws dt True
    let expectedNumEvents = n - 2 + 2 -- +1 for the sync event, +1 for the client add event
    preExistingEvents <- consumeEventsUntilEndOfInitialSync ws endMarker
    otherEvents <- consumeAllEvents ws
    let events = preExistingEvents <> otherEvents
    length events `shouldMatchInt` expectedNumEvents

mkEvent :: (ToJSON a1, ToJSON a2) => a1 -> a2 -> Bool -> Value
mkEvent uid cid transient =
  object
    [ "recipients" .= [object ["user_id" .= uid, "clients" .= [cid], "route" .= "any"]],
      "payload" .= [object ["hello" .= "world", "type" .= "test"]],
      "transient" .= transient
    ]

testTypingIndicatorIsNotSentToOwnClient :: (HasCallStack) => TaggedBool "federated" -> App ()
testTypingIndicatorIsNotSentToOwnClient (TaggedBool federated) = do
  let runTest =
        if federated
          then startDynamicBackends [(enableConsumableNotifications def), (enableConsumableNotifications def)]
          else \run -> startDynamicBackends [(enableConsumableNotifications def)] $ \[domain] -> run [domain, domain]
  runTest $ \[domain, otherDomain] -> do
    (alice, _, aliceClient) <- mkUserPlusClientWithDomain domain
    (bob, _, bobClient) <- mkUserPlusClientWithDomain otherDomain
    connectTwoUsers alice bob
    aliceClientId <- objId aliceClient
    bobClientId <- objId bobClient
    conv <- postConversation alice defProteus {qualifiedUsers = [bob]} >>= getJSON 201

    runCodensity (createEventWebSockets [(alice, Just aliceClientId), (bob, Just bobClientId)]) $ \[aliceWs, bobWs] -> do
      -- consume all events to ensure we start with a clean slate
      consumeAllEvents_ aliceWs
      consumeAllEvents_ bobWs

      -- Alice is typing
      sendTypingStatus alice conv "started" >>= assertSuccess

      -- Bob should receive the typing indicator for Alice
      assertEvent bobWs $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "conversation.typing"
        e %. "data.event.payload.0.qualified_conversation" `shouldMatch` (conv %. "qualified_id")
        e %. "data.event.payload.0.qualified_from" `shouldMatch` (alice %. "qualified_id")
        ackEvent bobWs e

      -- Alice should not receive the typing indicator for herself
      assertNoEvent_ aliceWs

      -- Bob is typing
      sendTypingStatus bob conv "started" >>= assertSuccess

      -- Alice should receive the typing indicator for Bob
      assertEvent aliceWs $ \e -> do
        e %. "data.event.payload.0.type" `shouldMatch` "conversation.typing"
        e %. "data.event.payload.0.qualified_conversation" `shouldMatch` (conv %. "qualified_id")
        e %. "data.event.payload.0.qualified_from" `shouldMatch` (bob %. "qualified_id")
        ackEvent aliceWs e

      -- Bob should not receive the typing indicator for himself
      assertNoEvent_ bobWs

-- convert :: ((HasCallStack) => (String -> App ()) -> App ()) -> ([String] -> App ()) -> App ()
-- convert = undefined

-- We only delete queues to clean up federated integration tests. So, we
-- mostly want to ensure we don't get stuck there.
testBackendPusherRecoversFromQueueDeletion :: (HasCallStack) => App ()
testBackendPusherRecoversFromQueueDeletion = do
  bob <- randomUser OwnDomain def

  domain1 <- asks (.domain1)

  let remotesRefreshInterval = 10000 :: Int
  startDynamicBackendsReturnResources
    [ enableConsumableNotifications
        ( def
            { backgroundWorkerCfg =
                setField
                  "backendNotificationPusher.remotesRefreshInterval"
                  remotesRefreshInterval
            }
        )
    ]
    $ \[beResource] -> do
      let domain = beResource.berDomain
      (alice, team, [alex, alison]) <- createTeam domain 3

      -- Create a federated conversion
      connectTwoUsers alice bob
      [alexId, alisonId, bobId] <-
        forM [alex, alison, bob] (%. "qualified_id")
      let nc = (defProteus {qualifiedUsers = [alexId, alisonId, bobId], team = Just team})
      void $ postConversation alice nc >>= getJSON 201

      withWebSockets [bob] $ \[wsBob] -> do
        rabbitmqAdminClient <- mkRabbitMqAdminClientForResource beResource
        let getActiveQueues :: App [String] =
              Text.unpack . (.name)
                <$$> ( (.items)
                         <$> rabbitmqAdminClient.listQueuesByVHost
                           (fromString beResource.berVHost)
                           (fromString "")
                           True
                           100
                           1
                     )

        void $ deleteTeamMember team alice alex >>= getBody 202

        assertConvUserDeletedNotif wsBob alexId

        -- Delete the queue
        let backendNotificationQueueName = "backend-notifications." <> domain1
        void
          $ rabbitmqAdminClient.deleteQueue
            (fromString beResource.berVHost)
            (fromString backendNotificationQueueName)

        -- Ensure the queue was deleted
        eventually $ do
          queueNames <- getActiveQueues
          queueNames `shouldNotContain` [backendNotificationQueueName]

        void $ deleteTeamMember team alice alison >>= getBody 202

        Timeout.threadDelay . fromIntegral $ remotesRefreshInterval * 2

        -- Check that the queue was recreated
        eventually $ do
          queueNames <- getActiveQueues
          queueNames `shouldContain` [backendNotificationQueueName]

        assertConvUserDeletedNotif wsBob alisonId

----------------------------------------------------------------------
-- helpers
mkUserPlusClientWithDomain :: (HasCallStack, MakesValue domain) => domain -> App (Value, String, String)
mkUserPlusClientWithDomain domain = do
  user <- randomUser domain def
  uid <- objId user
  client <- addClient user def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
  cid <- objId client
  pure (user, uid, cid)

mkUserPlusClient :: (HasCallStack) => App (Value, String, String)
mkUserPlusClient = mkUserPlusClientWithDomain OwnDomain

data EventWebSocket = EventWebSocket
  { events :: Chan (Either WS.ConnectionException Value),
    ack :: MVar (Maybe Value),
    kill :: MVar (),
    done :: MVar ()
  }

createEventWebSockets ::
  (HasCallStack, MakesValue user) =>
  [(user, Maybe String)] ->
  Codensity App [EventWebSocket]
createEventWebSockets = traverse (uncurry createEventsWebSocket)

createEventsWebSocket ::
  (HasCallStack, MakesValue user) =>
  user ->
  Maybe String ->
  Codensity App EventWebSocket
createEventsWebSocket user cid = do
  eithWS <- createEventsWebSocketEither user cid Nothing
  case eithWS of
    Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e
    Right ws -> pure ws

createEventsWebSocketWithSync ::
  (HasCallStack, MakesValue user) =>
  user ->
  Maybe String ->
  Codensity App (String, EventWebSocket)
createEventsWebSocketWithSync user cid = do
  syncMarker <- lift randomId
  eithWS <- createEventsWebSocketEither user cid (Just syncMarker)
  case eithWS of
    Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e
    Right ws -> pure (syncMarker, ws)

createEventsWebSocketEither ::
  (HasCallStack, MakesValue user) =>
  user ->
  Maybe String ->
  Maybe String ->
  Codensity App (Either WS.HandshakeException EventWebSocket)
createEventsWebSocketEither user cid mSyncMarker = do
  eventsChan <- liftIO newChan
  ackChan <- liftIO newEmptyMVar
  serviceMap <- lift $ getServiceMap =<< objDomain user
  apiVersion <- lift $ getAPIVersionFor $ objDomain user
  wsStarted <- newEmptyMVar
  let minAPIVersion = 8
  lift
    . when (apiVersion < minAPIVersion)
    $ assertFailure ("Events websocket can only be created when APIVersion is at least " <> show minAPIVersion)

  varKill <- lift $ newEmptyMVar
  varDone <- lift $ newEmptyMVar

  uid <- lift $ objId =<< objQidObject user
  let HostPort caHost caPort = serviceHostPort serviceMap Cannon
      path = "/v" <> show apiVersion <> "/events" <> maybe "" ("?client=" <>) cid <> maybe "" ("&sync_marker=" <>) mSyncMarker
      caHdrs = [(fromString "Z-User", toByteString' uid)]
      app conn = do
        putMVar wsStarted (Right ())
        race_
          (wsRead conn `catch` (writeChan eventsChan . Left))
          (wsWrite conn)

      wsRead conn = withAsync (wsReadLoop conn) $ \loop -> do
        r <- race (takeMVar varKill) (wait loop)
        case r of
          Left _ -> cancel loop >> waitClosed conn
          Right _ -> pure ()

      waitClosed conn = do
        WS.receive conn >>= \case
          WS.ControlMessage (WS.Close _ _) ->
            void $ tryPutMVar varDone ()
          _ -> waitClosed conn

      wsReadLoop conn = forever $ do
        bs <- WS.receiveData conn
        case decodeStrict' bs of
          Just n -> writeChan eventsChan (Right n)
          Nothing ->
            error $ "Failed to decode events: " ++ show bs

      wsWrite conn = do
        mAck <- takeMVar ackChan
        case mAck of
          Nothing -> WS.sendClose conn (Text.pack "")
          Just ack ->
            WS.sendBinaryData conn (encode ack)
              >> wsWrite conn

  wsThread <-
    Codensity
      $ withAsync
      $ liftIO
      $ WS.runClientWith caHost (fromIntegral caPort) path WS.defaultConnectionOptions caHdrs app
      `catch` \(e :: WS.HandshakeException) -> putMVar wsStarted (Left e)

  timeOutSeconds <- asks (.timeOutSeconds)
  mStarted <- lift $ timeout (timeOutSeconds * 1_000_000) (takeMVar wsStarted)
  case mStarted of
    Nothing -> do
      cancel wsThread
      lift $ assertFailure $ "Websocket failed to connect within " <> show timeOutSeconds <> "s"
    Just (Left e) ->
      pure (Left e)
    Just (Right ()) ->
      Codensity $ \k ->
        k (Right $ EventWebSocket eventsChan ackChan varKill varDone) `finally` do
          putMVar ackChan Nothing
          liftIO $ wait wsThread

ackFullSync :: (HasCallStack) => EventWebSocket -> App ()
ackFullSync ws =
  putMVar ws.ack
    $ Just (object ["type" .= "ack_full_sync"])

ackEvent :: (HasCallStack) => EventWebSocket -> Value -> App ()
ackEvent ws event = do
  deliveryTag <- event %. "data.delivery_tag"
  sendAck ws deliveryTag False

sendAck :: (HasCallStack) => EventWebSocket -> Value -> Bool -> App ()
sendAck ws deliveryTag multiple =
  do
    putMVar $ ws.ack
    $ Just
    $ object
      [ "type" .= "ack",
        "data"
          .= object
            [ "delivery_tag" .= deliveryTag,
              "multiple" .= multiple
            ]
      ]

killWebSocketClient :: EventWebSocket -> App ()
killWebSocketClient ws = do
  void $ tryPutMVar ws.kill ()
  takeMVar ws.done

assertEvent :: (HasCallStack) => EventWebSocket -> ((HasCallStack) => Value -> App a) -> App a
assertEvent ws expectations = do
  timeOutSeconds <- asks (.timeOutSeconds)
  timeout (timeOutSeconds * 1_000_000) (readChan ws.events) >>= \case
    Nothing -> assertFailure $ "No event received for " <> show timeOutSeconds <> "s"
    Just (Left ex) ->
      addFailureContext ("WSException: " <> displayException ex)
        $ assertFailure "Websocket closed when waiting for more events"
    Just (Right e) -> do
      pretty <- prettyJSON e
      addFailureContext ("event:\n" <> pretty)
        $ expectations e

-- | Tolerates and consumes other events before expected event
assertFindsEvent :: forall a. (HasCallStack) => EventWebSocket -> ((HasCallStack) => Value -> App a) -> App a
assertFindsEvent = assertFindsEventConfigurableAck ackEvent

assertFindsEventConfigurableAck ::
  forall a.
  (HasCallStack) =>
  ((HasCallStack) => EventWebSocket -> Value -> App ()) ->
  EventWebSocket ->
  ((HasCallStack) => Value -> App a) ->
  App a
assertFindsEventConfigurableAck ackFun ws expectations = go 0
  where
    go :: Int -> App a
    go ignoredEventCount = do
      timeOutSeconds <- asks (.timeOutSeconds)
      timeout (timeOutSeconds * 1_000_000) (readChan ws.events) >>= \case
        Nothing -> assertFailure $ show ignoredEventCount <> " event(s) received, no matching event received for " <> show timeOutSeconds <> "s"
        Just (Left ex) ->
          addFailureContext ("WSException: " <> displayException ex)
            $ assertFailure "Websocket closed when waiting for more events"
        Just (Right ev) -> do
          (expectations ev)
            `catch` \(_ :: AssertionFailure) -> do
              ignoredEventType <-
                maybe (pure "No Type") asString
                  =<< runMaybeT
                    ( (lookupFieldM ev "data.event" >>= flip lookupFieldM "payload.0.type")
                        <|> (lookupFieldM ev "type")
                    )
              ackFun ws ev
              addJSONToFailureContext ("Ignored Event (" <> ignoredEventType <> ")") ev
                $ go (ignoredEventCount + 1)

data NoEvent = NoEvent | WebSocketDied

instance ToJSON NoEvent where
  toJSON NoEvent = toJSON "no-event"
  toJSON WebSocketDied = toJSON "web-socket-died"

assertNoEventHelper :: (HasCallStack) => EventWebSocket -> App NoEvent
assertNoEventHelper ws = do
  timeOutSeconds <- asks (.timeOutSeconds)
  timeout (timeOutSeconds * 1_000_000) (readChan ws.events) >>= \case
    Nothing -> pure NoEvent
    Just (Left _) -> pure WebSocketDied
    Just (Right e) -> do
      eventJSON <- prettyJSON e
      assertFailure $ "Did not expect event: \n" <> eventJSON

-- | Similar to `assertNoEvent` from Testlib, but with rabbitMQ typing (`/event` end-point, not
-- `/await`).
assertNoEvent_ :: (HasCallStack) => EventWebSocket -> App ()
assertNoEvent_ = void . assertNoEventHelper

assertWebSocketDied :: (HasCallStack) => EventWebSocket -> App ()
assertWebSocketDied ws = do
  recpol <- do
    timeOutSeconds <- asks (.timeOutSeconds)
    pure $ limitRetriesByCumulativeDelay (timeOutSeconds * 1_000_000) (constantDelay 800_000)
  recoverAll recpol $ \_ ->
    assertNoEventHelper ws >>= \case
      NoEvent -> assertFailure $ "WebSocket is still open"
      WebSocketDied -> pure ()

consumeEventsUntilEndOfInitialSync :: (HasCallStack) => EventWebSocket -> String -> App [Value]
consumeEventsUntilEndOfInitialSync ws expectedMarkerId = go []
  where
    go events = do
      timeout 1_000_000 (readChan ws.events) >>= \case
        Nothing ->
          addJSONToFailureContext "events" events
            $ assertFailure "timed out waiting for end-of-initial-sync event"
        Just (Left e) ->
          assertFailure
            $ "Websocket closed while waiting for end-of-initial-sync event "
            <> displayException e
        Just (Right e) -> do
          ackEvent ws e
          t <- e %. "type" & asString
          if (t == "synchronization")
            then do
              markerId <- e %. "data.marker_id" & asString
              if (markerId == expectedMarkerId)
                then pure (events <> [e])
                else assertFailure $ "Expected marker_id " <> expectedMarkerId <> ", but got " <> markerId
            else go (events <> [e])

assertEndOfIniitalSync :: (HasCallStack) => EventWebSocket -> String -> App ()
assertEndOfIniitalSync ws endMarker =
  assertEvent ws $ \e -> do
    e %. "type" `shouldMatch` "synchronization"
    e %. "data.marker_id" `shouldMatch` endMarker

consumeAllEvents_ :: EventWebSocket -> App ()
consumeAllEvents_ = void . consumeAllEvents

emptyQueue :: (HasCallStack, MakesValue user) => user -> String -> App ()
emptyQueue user cid = do
  runCodensity (createEventsWebSocketWithSync user (Just cid)) $ \(endMarker, ws) -> do
    void $ consumeEventsUntilEndOfInitialSync ws endMarker

consumeAllEvents :: EventWebSocket -> App [Value]
consumeAllEvents ws = do
  timeout 1_000_000 (readChan ws.events) >>= \case
    Nothing -> pure []
    Just (Left e) ->
      assertFailure
        $ "Websocket closed while consuming all events: "
        <> displayException e
    Just (Right e) -> do
      ackEvent ws e
      (e :) <$> consumeAllEvents ws

consumeAllEventsNoAck :: EventWebSocket -> App [Value]
consumeAllEventsNoAck ws = do
  timeout 1_000_000 (readChan ws.events) >>= \case
    Nothing -> pure []
    Just (Left e) ->
      assertFailure
        $ "Websocket closed while consuming all events: "
        <> displayException e
    Just (Right e) -> do
      (e :) <$> consumeAllEventsNoAck ws

-- | Only considers connections from cannon
waitUntilNoRabbitMqConns :: (HasCallStack) => BackendResource -> App ()
waitUntilNoRabbitMqConns backend = do
  rabbitmqAdminClient <- mkRabbitMqAdminClientForResource backend
  recoverAll
    (constantDelay 500_000 <> limitRetries 10)
    (const (go rabbitmqAdminClient))
  where
    go rabbitmqAdminClient = do
      cannonConnections <- getCannonConnections rabbitmqAdminClient backend.berVHost
      cannonConnections `shouldMatch` ([] :: [Connection])

-- | Only kills connections from cannon and returns them
killAllRabbitMqConns :: (HasCallStack) => BackendResource -> App [Connection]
killAllRabbitMqConns backend = do
  rabbitmqAdminClient <- mkRabbitMqAdminClientForResource backend
  cannonConnections <- getCannonConnections rabbitmqAdminClient backend.berVHost
  for_ cannonConnections $ \connection ->
    rabbitmqAdminClient.deleteConnection connection.name
  pure cannonConnections

getCannonConnections :: AdminAPI (AsClientT App) -> String -> App [Connection]
getCannonConnections rabbitmqAdminClient vhost = do
  connections <- rabbitmqAdminClient.listConnectionsByVHost (Text.pack vhost)
  pure $ filter (\c -> maybe False (fromString "pool " `Text.isPrefixOf`) c.userProvidedName) connections

mkRabbitMqAdminClientForResource :: BackendResource -> App (AdminAPI (Servant.AsClientT App))
mkRabbitMqAdminClientForResource backend = do
  opts <- asks (.rabbitMQConfig)
  servantClient <- liftIO $ mkRabbitMqAdminClientEnv opts {vHost = Text.pack backend.berVHost}
  pure . fromServant $ Servant.hoistClient (Proxy @(ToServant AdminAPI AsApi)) (liftIO @App) (toServant servantClient)

enableConsumableNotifications :: ServiceOverrides -> ServiceOverrides
enableConsumableNotifications overrides =
  overrides
    <> def
      { brigCfg = setField "optSettings.setConsumableNotifications" True,
        gundeckCfg = setField "settings.consumableNotifications" True
      }
