-- This file is part of the Wire Server implementation.
--
-- Copyright (C) 2022 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/>.
{-# LANGUAGE RecordWildCards #-}

-- 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 Gundeck.Push
  ( push,
    addToken,
    listTokens,
    deleteToken,
    -- (for testing)
    pushAll,
    splitPush,
    MonadPushAll (..),
    MonadNativeTargets (..),
    MonadMapAsync (..),
  )
where

import Bilge qualified
import Bilge.RPC qualified as Bilge
import Control.Error
import Control.Lens (to, view, (.~), (^.))
import Control.Monad.Catch
import Control.Monad.Except (throwError)
import Data.Aeson qualified as Aeson
import Data.ByteString.Conversion (toByteString')
import Data.Id
import Data.List.Extra qualified as List
import Data.List.NonEmpty (NonEmpty ((:|)))
import Data.Map qualified as Map
import Data.Misc
import Data.Set qualified as Set
import Data.Text qualified as Text
import Data.These
import Data.Timeout
import Data.UUID qualified as UUID
import Gundeck.Aws (endpointUsers)
import Gundeck.Aws qualified as Aws
import Gundeck.Aws.Arn
import Gundeck.Env
import Gundeck.Monad
import Gundeck.Notification.Data qualified as Data
import Gundeck.Options
import Gundeck.Presence.Data qualified as Presence
import Gundeck.Push.Data qualified as Data
import Gundeck.Push.Native qualified as Native
import Gundeck.Push.Native.Types
import Gundeck.Push.Websocket qualified as Web
import Gundeck.ThreadBudget
import Gundeck.Util
import Imports
import Network.AMQP (Message (..))
import Network.AMQP qualified as Q
import Network.HTTP.Types
import Network.Wai.Utilities
import System.Logger.Class (msg, val, (+++), (.=), (~~))
import System.Logger.Class qualified as Log
import UnliftIO (pooledMapConcurrentlyN)
import Util.Options
import Wire.API.Internal.Notification
import Wire.API.Notification
import Wire.API.Presence (Presence (..))
import Wire.API.Push.Token qualified as Public
import Wire.API.Push.V2
import Wire.API.User (UserSet (..))
import Wire.API.User.Client (Client (..), UserClientsFull (..), supportsConsumableNotifications)

push :: [Push] -> Gundeck ()
push ps = do
  pushAll ps `catch` \(ex :: SomeException) -> do
    Log.err $ msg (val "Push failed") . Log.field "error" (displayException ex)
    throwM (mkError status500 "server-error" "Server Error")

-- | Abstract over all effects in 'pushAll' (for unit testing).
class (MonadThrow m) => MonadPushAll m where
  mpaConsumableNotificationsEnabled :: m Bool
  mpaNotificationTTL :: m NotificationTTL
  mpaCellsEventQueue :: m (Maybe Text)
  mpaMkNotificationId :: m NotificationId
  mpaListAllPresences :: [UserId] -> m [[Presence]]
  mpaBulkPush :: [(Notification, [Presence])] -> m [(NotificationId, [Presence])]
  mpaStreamAdd :: NotificationId -> NonEmpty NotificationTarget -> NonEmpty Aeson.Object -> NotificationTTL -> m ()
  mpaPushNative :: Notification -> Priority -> [Address] -> m ()
  mpaForkIO :: m () -> m ()
  mpaRunWithBudget :: Int -> a -> m a -> m a
  mpaGetClients :: Set UserId -> m UserClientsFull
  mpaPublishToRabbitMq :: Text -> Text -> Q.Message -> m ()

instance MonadPushAll Gundeck where
  mpaConsumableNotificationsEnabled = view (options . settings . consumableNotifications)
  mpaNotificationTTL = view (options . settings . notificationTTL)
  mpaCellsEventQueue = view (options . settings . cellsEventQueue)
  mpaMkNotificationId = mkNotificationId
  mpaListAllPresences = runWithDefaultRedis . Presence.listAll
  mpaBulkPush = Web.bulkPush
  mpaStreamAdd = Data.add
  mpaPushNative = pushNative
  mpaForkIO = void . forkIO
  mpaRunWithBudget = runWithBudget''
  mpaGetClients = getClients
  mpaPublishToRabbitMq = publishToRabbitMq

publishToRabbitMq :: Text -> Text -> Q.Message -> Gundeck ()
publishToRabbitMq exchangeName routingKey qMsg = do
  chan <- getRabbitMqChan
  void $ liftIO $ Q.publishMsg chan exchangeName routingKey qMsg

-- | Another layer of wrap around 'runWithBudget'.
runWithBudget'' :: Int -> a -> Gundeck a -> Gundeck a
runWithBudget'' budget fallback action = do
  view threadBudgetState >>= \case
    Nothing -> action
    Just tbs -> runWithBudget' tbs budget fallback action

-- | Abstract over all effects in 'nativeTargets' (for unit testing).
class (Monad m) => MonadNativeTargets m where
  mntgtLogErr :: SomeException -> m ()
  mntgtLookupAddresses :: UserId -> m [Address]

instance MonadNativeTargets Gundeck where
  mntgtLogErr e = Log.err (msg (val "Failed to get native push address: " +++ show e))
  mntgtLookupAddresses rcp = Data.lookup rcp Data.One

class (Monad m) => MonadMapAsync m where
  mntgtMapAsync :: (a -> m b) -> [a] -> m [Either SomeException b]
  mntgtPerPushConcurrency :: m (Maybe Int)

instance MonadMapAsync Gundeck where
  mntgtPerPushConcurrency = view (options . settings . perNativePushConcurrency)
  mntgtMapAsync f l = do
    perPushConcurrency <- mntgtPerPushConcurrency
    case perPushConcurrency of
      Nothing -> mapAsync f l
      Just chunkSize -> concat <$> mapM (mapAsync f) (List.chunksOf chunkSize l)

splitPushes :: (MonadPushAll m) => [Push] -> m ([Push], [Push], UserClientsFull)
splitPushes ps = do
  allUserClients <- mpaGetClients (Set.unions $ map (\p -> Set.map (._recipientId) $ p._pushRecipients) ps)
  let (rabbitmqPushes, legacyPushes) = partitionHereThere $ map (splitPush allUserClients) ps
  pure (rabbitmqPushes, legacyPushes, allUserClients)

-- | Split a push into rabbitmq and legacy push. This code exists to help with
-- migration. Once it is completed and old APIs are not supported anymore we can
-- assume everything is meant for RabbitMQ and stop splitting.
splitPush ::
  UserClientsFull ->
  Push ->
  -- | These rabbitmqPush cassandraPush
  These Push Push
splitPush clientsFull p = do
  let (rabbitmqRecipients, legacyRecipients) =
        partitionHereThere $
          map splitRecipient (toList p._pushRecipients)
  case (null rabbitmqRecipients, null legacyRecipients) of
    (True, _) -> (That p)
    (_, True) -> (This p)
    (False, False) ->
      These
        p {_pushRecipients = Set.fromList rabbitmqRecipients}
        p {_pushRecipients = Set.fromList legacyRecipients}
  where
    splitRecipient :: Recipient -> These Recipient Recipient
    splitRecipient rcpt = do
      let allClients = Map.findWithDefault mempty rcpt._recipientId $ clientsFull.userClientsFull
          relevantClients = case rcpt._recipientClients of
            RecipientClientsSome cs ->
              Set.filter (\c -> c.clientId `elem` toList cs) allClients
            RecipientClientsAll -> allClients
          (rabbitmqClients, legacyClients) = Set.partition supportsConsumableNotifications relevantClients
          rabbitmqClientIds = (.clientId) <$> Set.toList rabbitmqClients
          legacyClientIds = (.clientId) <$> Set.toList legacyClients
      case (rabbitmqClientIds, legacyClientIds) of
        ([], _) ->
          -- Checking for rabbitmqClientIds first ensures that we fall back to
          -- old behaviour even if legacyClientIds is empty too. This way we
          -- won't break things before clients are ready for it.
          --
          -- We return all clients for RabbitMQ even if there are no real
          -- clients so a temporary client can still read the notifications on
          -- RabbitMQ.
          That rcpt
        (_, []) ->
          This rcpt
        (r : rs, l : ls) ->
          let rabbitMqRecipients = case rcpt._recipientClients of
                RecipientClientsAll -> RecipientClientsAll
                RecipientClientsSome _ -> RecipientClientsSome $ r :| rs
           in These
                rcpt {_recipientClients = rabbitMqRecipients}
                rcpt {_recipientClients = RecipientClientsSome $ l :| ls}

getClients :: Set UserId -> Gundeck UserClientsFull
getClients uids = do
  fmap mconcat
    . pooledMapConcurrentlyN 4 getBatch
    . List.chunksOf 100
    $ Set.toList uids
  where
    getBatch :: [UserId] -> Gundeck UserClientsFull
    getBatch uidsChunk = do
      r <- do
        Endpoint h p <- view $ options . brig
        Bilge.rpc "brig" $
          Bilge.method POST
            . Bilge.host (toByteString' h)
            . Bilge.port p
            . Bilge.path "/i/clients/full"
            . Bilge.json (UserSet $ Set.fromList uidsChunk)
            . Bilge.expect2xx
      Bilge.responseJsonError r

pushAll :: (MonadPushAll m, MonadNativeTargets m, MonadMapAsync m, Log.MonadLogger m) => [Push] -> m ()
pushAll pushes = do
  Log.debug $ msg (val "pushing") . Log.field "pushes" (Aeson.encode pushes)
  consumableNotificationsEnabled <- mpaConsumableNotificationsEnabled
  (rabbitmqPushes, legacyPushes, allUserClients) <-
    if consumableNotificationsEnabled
      then splitPushes pushes
      else do
        allUserClients <- mpaGetClients (Set.unions $ map (\p -> Set.map (._recipientId) $ p._pushRecipients) pushes)
        pure ([], pushes, allUserClients)

  legacyNotifs <- mapM mkNewNotification legacyPushes
  pushAllLegacy legacyNotifs allUserClients

  rabbitmqNotifs <- mapM mkNewNotification rabbitmqPushes
  unless (null rabbitmqNotifs) $ pushAllViaRabbitMq rabbitmqNotifs allUserClients

  -- Note that Cells needs all notifications because it doesn't matter whether
  -- some recipients have rabbitmq clients or not.
  --
  -- We cannot simply merge 'legacyNotifs' and 'rabbitmqNotifs' because
  -- notifications could be duplicated in case cells notifications have users
  -- who have legacy and rabbitmq clients.
  allNotifs <- mapM mkNewNotification pushes
  pushAllToCells allNotifs

-- | Construct and send a single bulk push request to the client.  Write the 'Notification's from
-- the request to C*.  Trigger native pushes for all delivery failures notifications.
pushAllLegacy :: (MonadPushAll m, MonadNativeTargets m, MonadMapAsync m) => [NewNotification] -> UserClientsFull -> m ()
pushAllLegacy newNotifications userClientsFull = do
  -- persist push request
  let cassandraTargets :: [CassandraTargets]
      cassandraTargets = map mkCassandraTargets newNotifications
  forM_ cassandraTargets $ \CassandraTargets {..} ->
    unless (ntfTransient ctNotification) $
      case ctNotificationTargets of
        [] -> pure ()
        t : ts ->
          mpaStreamAdd (ntfId ctNotification) (t :| ts) (ntfPayload ctNotification)
            =<< mpaNotificationTTL
  mpaForkIO $ do
    -- websockets
    wsTargets <- mapM mkWSTargets newNotifications
    resp <- compilePushResps wsTargets <$> mpaBulkPush (compilePushReq <$> wsTargets)
    -- native push
    forM_ resp $ \((notif :: Notification, psh :: Push), alreadySent :: [Presence]) -> do
      let alreadySentClients = Set.fromList $ mapMaybe (\p -> (p.userId,) <$> p.clientId) alreadySent
          rabbitmqClients = Map.map (Set.filter supportsConsumableNotifications) userClientsFull.userClientsFull
          rabbitmqClientIds = Map.foldMapWithKey (\uid clients -> Set.map (\c -> (uid, c.clientId)) clients) rabbitmqClients
      pushNativeWithBudget notif psh (Set.toList $ Set.union alreadySentClients rabbitmqClientIds)

pushNativeWithBudget :: (MonadMapAsync m, MonadPushAll m, MonadNativeTargets m) => Notification -> Push -> [(UserId, ClientId)] -> m ()
pushNativeWithBudget notif psh dontPush = do
  perPushConcurrency <- mntgtPerPushConcurrency
  let rcps' = nativeTargetsRecipients psh
      cost = maybe (length rcps') (min (length rcps')) perPushConcurrency
  -- this is a rough budget cost, since there may be more than one device in a
  -- 'Presence', so one budget token may trigger at most 8 push notifications
  -- to be sent out.
  -- If perPushConcurrency is defined, we take the min with 'perNativePushConcurrency', as native push requests
  -- to cassandra and SNS are limited to 'perNativePushConcurrency' in parallel.
  unless (psh ^. pushTransient) $
    mpaRunWithBudget cost () $
      mpaPushNative notif (psh ^. pushNativePriority) =<< nativeTargets psh rcps' dontPush

pushAllViaRabbitMq :: (MonadPushAll m, MonadMapAsync m, MonadNativeTargets m) => [NewNotification] -> UserClientsFull -> m ()
pushAllViaRabbitMq newNotifs userClientsFull = do
  for_ newNotifs $ pushViaRabbitMq
  mpaForkIO $ do
    for_ newNotifs $ \newNotif -> do
      let cassandraClients = Map.map (Set.filter $ not . supportsConsumableNotifications) userClientsFull.userClientsFull
          cassandraClientIds = Map.foldMapWithKey (\uid clients -> Set.map (\c -> (uid, c.clientId)) clients) cassandraClients
      pushNativeWithBudget newNotif.nnNotification newNotif.nnPush (Set.toList $ cassandraClientIds)

pushViaRabbitMq :: (MonadPushAll m) => NewNotification -> m ()
pushViaRabbitMq newNotif = do
  qMsg <- mkMessage newNotif.nnNotification
  let routingKeys =
        Set.unions $
          flip Set.map (Set.fromList . toList $ newNotif.nnRecipients) \r ->
            case r._recipientClients of
              RecipientClientsAll ->
                Set.singleton $ userRoutingKey r._recipientId
              RecipientClientsSome (toList -> cs) ->
                Set.fromList $ map (clientRoutingKey r._recipientId) cs
  for_ routingKeys $ \routingKey ->
    mpaPublishToRabbitMq userNotificationExchangeName routingKey qMsg

pushAllToCells :: (MonadPushAll m, Log.MonadLogger m) => [NewNotification] -> m ()
pushAllToCells newNotifs = do
  mQueue <- mpaCellsEventQueue

  let cellsNotifs = filter (.nnPush._pushIsCellsEvent) newNotifs
  unless (null cellsNotifs) $ case mQueue of
    Nothing -> do
      Log.err $ msg ("Cells events have been generated, but no Cells queue is configured in gundeck" :: ByteString)
    Just queue ->
      traverse_ (pushToCells queue) cellsNotifs

pushToCells :: (MonadPushAll m) => Text -> NewNotification -> m ()
pushToCells queue newNotif = do
  qMsg <- mkMessage newNotif.nnNotification
  mpaPublishToRabbitMq "" queue qMsg

mkMessage :: (MonadPushAll m) => Notification -> m Message
mkMessage notif = do
  NotificationTTL ttl <- mpaNotificationTTL
  pure $
    Q.newMsg
      { msgBody =
          Aeson.encode
            . queuedNotification notif.ntfId
            $ notif.ntfPayload,
        msgContentType = Just "application/json",
        msgDeliveryMode =
          -- Non-persistent messages never hit the disk and so do not
          -- survive RabbitMQ node restarts, this is great for transient
          -- notifications.
          Just
            ( if notif.ntfTransient
                then Q.NonPersistent
                else Q.Persistent
            ),
        msgExpiration =
          Just
            ( if notif.ntfTransient
                then
                  ( -- Means that if there is no active consumer, this
                    -- message will never be delivered to anyone. It can
                    -- still take some time before RabbitMQ forgets about
                    -- this message because the expiration is only
                    -- considered for messages which are at the head of a
                    -- queue. See docs: https://www.rabbitmq.com/docs/ttl
                    "0"
                  )
                else showT $ fromIntegral ttl # Second #> MilliSecond
            )
      }

-- | A new notification to be stored in C* and pushed over websockets
data NewNotification = NewNotification
  { nnPush :: Push,
    nnNotification :: Notification,
    nnRecipients :: [Recipient]
  }

mkNewNotification :: forall m. (MonadPushAll m) => Push -> m NewNotification
mkNewNotification psh = do
  let rcps = toList $ psh ^. pushRecipients
  notifId <- mpaMkNotificationId
  let notif = Notification notifId (psh ^. pushTransient) (psh ^. pushPayload)
  pure $ NewNotification psh notif rcps

-- | Information to be stored in cassandra for every 'Notification'.
data CassandraTargets = CassandraTargets
  { ctNotification :: Notification,
    ctNotificationTargets :: [NotificationTarget]
  }

mkCassandraTargets :: NewNotification -> CassandraTargets
mkCassandraTargets NewNotification {..} =
  CassandraTargets nnNotification (mapMaybe mkTarget $ toList nnRecipients)
  where
    mkTarget :: Recipient -> Maybe NotificationTarget
    mkTarget r = do
      clients <-
        case r ^. recipientClients of
          RecipientClientsAll -> Just []
          -- clients are stored in cassandra as a list with a notification.  empty list
          -- is interpreted as "all clients" by 'Gundeck.Notification.Data.toNotif'.
          RecipientClientsSome cs -> Just $ toList cs
      pure $ target (r ^. recipientId) & targetClients .~ clients

-- | Information needed to push notifications over websockets and/or native
-- pushes.
data WSTargets = WSTargets
  { wstPush :: Push,
    wstNotification :: Notification,
    wstPresences :: [(Recipient, [Presence])]
  }

mkWSTargets :: (MonadPushAll m) => NewNotification -> m WSTargets
mkWSTargets NewNotification {..} = do
  withPresences <- addPresences nnRecipients
  pure $ WSTargets nnPush nnNotification withPresences
  where
    addPresences :: forall m. (MonadPushAll m) => [Recipient] -> m [(Recipient, [Presence])]
    addPresences rcps = do
      presences <- mpaListAllPresences $ fmap (view recipientId) rcps
      pure $ zip rcps presences

-- REFACTOR: @[Presence]@ here should be @newtype WebSockedDelivered = WebSockedDelivered [Presence]@
compilePushReq :: WSTargets -> (Notification, [Presence])
compilePushReq WSTargets {..} =
  (wstNotification, mconcat . fmap compileTargets $ wstPresences)
  where
    compileTargets :: (Recipient, [Presence]) -> [Presence]
    compileTargets (rcp, pre) = filter (shouldActuallyPush wstPush rcp) pre

compilePushResps ::
  [WSTargets] ->
  [(NotificationId, [Presence])] ->
  [((Notification, Push), [Presence])]
compilePushResps notifIdMap (Map.fromList -> deliveries) =
  notifIdMap
    <&> (\WSTargets {..} -> ((wstNotification, wstPush), fromMaybe [] (Map.lookup (ntfId wstNotification) deliveries)))

-- | Is 'PushTarget' the origin of the 'Push', or is missing in a non-empty whitelist?  (Whitelists
-- reside both in 'Push' itself and in each 'Recipient').
shouldActuallyPush :: Push -> Recipient -> Presence -> Bool
shouldActuallyPush psh rcp pres = not isOrigin && okByRecipientAllowlist
  where
    isOrigin =
      psh ^. pushOrigin == Just (userId pres)
        && psh ^. pushOriginConnection == Just (connId pres)

    okByRecipientAllowlist :: Bool
    okByRecipientAllowlist =
      case (rcp ^. recipientClients, pres.clientId) of
        (RecipientClientsSome cs, Just c) -> c `elem` cs
        _ -> True

-- | Failures to push natively can be ignored.  Logging already happens in
-- 'Gundeck.Push.Native.push1', and we cannot recover from any of the error cases.
pushNative :: Notification -> Priority -> [Address] -> Gundeck ()
pushNative _ _ [] = pure ()
pushNative notif prio rcps = do
  Native.push (Native.NativePush (ntfId notif) prio Nothing) rcps

-- | Compute list of 'Recipient's from a 'Push' that may be interested in a native push.  More
-- filtering in 'nativeTargets'.
nativeTargetsRecipients :: Push -> [Recipient]
nativeTargetsRecipients psh = filter routeNative (toList (psh ^. pushRecipients))
  where
    routeNative u =
      u ^. recipientRoute /= RouteDirect
        && (Just (u ^. recipientId) /= psh ^. pushOrigin || psh ^. pushNativeIncludeOrigin)

-- | FUTUREWORK: 'nativeTargets' calls cassandra once for each 'Recipient' of the 'Push'.  Instead,
-- it should be called once with @[Push]@ for every call to 'pushAll', and that call should
-- only call cassandra once in total, yielding all addresses of all recipients of all pushes.
nativeTargets ::
  forall m.
  (MonadNativeTargets m, MonadMapAsync m) =>
  Push ->
  [Recipient] ->
  [(UserId, ClientId)] ->
  m [Address]
nativeTargets psh rcps' dontPush =
  mntgtMapAsync addresses rcps' >>= fmap concat . mapM check
  where
    addresses :: Recipient -> m [Address]
    addresses u = do
      addrs <- mntgtLookupAddresses (u ^. recipientId)
      pure $ filter (eligible u) addrs
    eligible :: Recipient -> Address -> Bool
    eligible u a
      -- Never include the origin client.
      | Just (a ^. addrUser) == psh ^. pushOrigin && Just (a ^. addrConn) == psh ^. pushOriginConnection = False
      -- Is the specific client an intended recipient?
      | not (eligibleClient a (u ^. recipientClients)) = False
      -- Include client if not found in already served presences.
      | otherwise = not $ List.elem (a ^. addrUser, a ^. addrClient) dontPush --  (List.find (isOnline a) alreadySent)
    eligibleClient _ RecipientClientsAll = True
    eligibleClient a (RecipientClientsSome cs) = (a ^. addrClient) `elem` cs

    check :: Either SomeException [a] -> m [a]
    check (Left e) = mntgtLogErr e >> pure []
    check (Right r) = pure r

type AddTokenResponse = Either Public.AddTokenError Public.AddTokenSuccess

addToken :: UserId -> ConnId -> PushToken -> Gundeck AddTokenResponse
addToken uid cid newtok = mpaRunWithBudget 1 (Left Public.AddTokenErrorNoBudget) $ runExceptT $ do
  when (newtok ^. tokenTransport `elem` [APNSVoIP, APNSVoIPSandbox]) $
    throwError Public.AddTokenErrorApnsVoipNotSupported

  (cur, old) <- lift $ foldl' (matching newtok) (Nothing, []) <$> Data.lookup uid Data.LocalQuorum
  lift $
    Log.info $
      "user"
        .= UUID.toASCIIBytes (toUUID uid)
        ~~ "token"
        .= Text.take 16 (tokenText (newtok ^. token))
        ~~ msg (val "Registering push token")
  addr <- continue newtok cur
  lift $ Native.deleteTokens old (Just addr)
  pure $ Public.AddTokenSuccess newtok
  where
    matching ::
      PushToken ->
      (Maybe Address, [Address]) ->
      Address ->
      (Maybe Address, [Address])
    matching t (x, old) a
      | a ^. addrTransport == t ^. tokenTransport
          && a ^. addrApp == t ^. tokenApp
          && a ^. addrClient == t ^. tokenClient =
          if a ^. addrToken == t ^. token
            then (Just a, old)
            else (x, a : old)
      | otherwise = (x, old)

    continue ::
      PushToken ->
      Maybe Address ->
      ExceptT Public.AddTokenError Gundeck Address
    continue t Nothing = create (0 :: Int) t
    continue t (Just a) = update (0 :: Int) t (a ^. addrEndpoint)

    create ::
      Int ->
      PushToken ->
      ExceptT Public.AddTokenError Gundeck Address
    create n t = do
      let trp = t ^. tokenTransport
      let app = t ^. tokenApp
      let tok = t ^. token
      env <- view (options . aws . arnEnv)
      aws' <- view awsEnv
      ept <- Aws.execute aws' (Aws.createEndpoint uid trp env app tok)
      case ept of
        Left (Aws.EndpointInUse arn) -> do
          lift $ Log.info $ "arn" .= toText arn ~~ msg (val "ARN in use")
          update (n + 1) t arn
        Left (Aws.AppNotFound app') -> do
          lift $ Log.info $ msg ("Push token of unknown application: '" <> appNameText app' <> "'")
          throwError Public.AddTokenErrorNotFound
        Left (Aws.InvalidToken _) -> do
          lift $
            Log.info $
              "token"
                .= tokenText tok
                ~~ msg (val "Invalid push token.")
          throwError Public.AddTokenErrorInvalid
        Left (Aws.TokenTooLong l) -> do
          lift $ Log.info $ msg ("Push token is too long: token length = " ++ show l)
          throwError Public.AddTokenErrorTooLong
        Right arn -> do
          Data.insert uid trp app tok arn cid (t ^. tokenClient)
          pure $ mkAddr t arn

    update ::
      Int ->
      PushToken ->
      SnsArn EndpointTopic ->
      ExceptT Public.AddTokenError Gundeck Address
    update n t arn = do
      when (n >= 3) $ do
        lift $ Log.err $ msg (val "AWS SNS inconsistency w.r.t. " +++ toText arn)
        throwM (mkError status500 "server-error" "Server Error")
      aws' <- view awsEnv
      ept <- Aws.execute aws' (Aws.lookupEndpoint arn)
      case ept of
        Nothing -> create (n + 1) t
        Just ep ->
          do
            lift $ updateEndpoint uid t arn ep
            Data.insert
              uid
              (t ^. tokenTransport)
              (t ^. tokenApp)
              (t ^. token)
              arn
              cid
              (t ^. tokenClient)
            pure $ mkAddr t arn
            `catch` \case
              -- Note: If the endpoint was recently deleted (not necessarily
              -- concurrently), we may get an EndpointNotFound error despite
              -- the previous lookup, i.e. endpoint lookups may exhibit eventually
              -- consistent semantics with regards to endpoint deletion (or
              -- possibly updates in general). We make another attempt to (re-)create
              -- the endpoint in these cases instead of failing immediately.
              Aws.EndpointNotFound {} -> create (n + 1) t
              Aws.InvalidCustomData {} -> throwError Public.AddTokenErrorMetadataTooLong
              ex -> throwM ex

    mkAddr ::
      PushToken ->
      EndpointArn ->
      Address
    mkAddr t arn =
      Address
        uid
        arn
        cid
        (pushToken (t ^. tokenTransport) (t ^. tokenApp) (t ^. token) (t ^. tokenClient))

-- | Update an SNS endpoint with the given user and token.
updateEndpoint :: UserId -> PushToken -> EndpointArn -> Aws.SNSEndpoint -> Gundeck ()
updateEndpoint uid t arn e = do
  env <- view awsEnv
  requestId <- view reqId

  unless (equalTransport && equalApp) $ do
    Log.err $ logMessage requestId "PushToken does not fit to user_push data: Transport or app mismatch"
    throwM $ mkError status500 "server-error" "Server Error"

  Log.info $ logMessage requestId "Upserting push token."
  let users = Set.insert uid (e ^. endpointUsers)
  Aws.execute env $ Aws.updateEndpoint users (t ^. token) arn
  where
    equalTransport = t ^. tokenTransport == arn ^. snsTopic . endpointTransport
    equalApp = t ^. tokenApp == arn ^. snsTopic . endpointAppName
    logMessage requestId m =
      "user"
        .= UUID.toASCIIBytes (toUUID uid)
        ~~ "token"
        .= Text.take 16 (t ^. token . to tokenText)
        ~~ "tokenTransport"
        .= show (t ^. tokenTransport)
        ~~ "tokenApp"
        .= (t ^. tokenApp . to appNameText)
        ~~ "arn"
        .= toText arn
        ~~ "endpointTransport"
        .= show (arn ^. snsTopic . endpointTransport)
        ~~ "endpointAppName"
        .= (arn ^. snsTopic . endpointAppName . to appNameText)
        ~~ "request"
        .= unRequestId requestId
        ~~ msg (val m)

deleteToken :: UserId -> Token -> Gundeck (Maybe ())
deleteToken uid tok = do
  Data.lookup uid Data.LocalQuorum
    >>= ( \case
            [] -> pure Nothing
            xs -> Native.deleteTokens xs Nothing $> Just ()
        )
      . filter (\x -> x ^. addrToken == tok)

listTokens :: UserId -> Gundeck PushTokenList
listTokens uid = PushTokenList . map (^. addrPushToken) <$> Data.lookup uid Data.LocalQuorum
