-- 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/>.

module Brig.Index.Eval
  ( runCommand,
    initIndex,
  )
where

import Brig.App (initHttpManagerWithTLSConfig, mkIndexEnv)
import Brig.Index.Options as IxOpts
import Brig.Options as Opt
import Brig.User.Search.Index
import Cassandra (ClientState)
import Cassandra.Options
import Cassandra.Util (defInitCassandra)
import Control.Exception (throwIO)
import Control.Lens
import Control.Monad.Catch
import Control.Retry
import Data.Aeson (FromJSON)
import Data.Aeson qualified as Aeson
import Data.ByteString.Lazy.UTF8 qualified as UTF8
import Data.Credentials (Credentials (..))
import Data.Id
import Database.Bloodhound qualified as ES
import Database.Bloodhound.Internal.Client (BHEnv (..))
import Hasql.Pool (UsageError)
import Hasql.Pool.Extended
import Hasql.Pool.Extended qualified as Hasql
import Imports
import Network.HTTP.Client (Manager)
import Polysemy
import Polysemy.Error
import Polysemy.Input
import Polysemy.TinyLog (TinyLog)
import System.Logger qualified as Log
import System.Logger.Class (Logger)
import Util.Options
import Wire.ClientSubsystem.Error (ClientError)
import Wire.GalleyAPIAccess (GalleyAPIAccess)
import Wire.GalleyAPIAccess.Rpc
import Wire.IndexedUserStore
import Wire.IndexedUserStore.Bulk.ElasticSearch qualified as IndexedUserStoreBulk
import Wire.IndexedUserStore.ElasticSearch
import Wire.IndexedUserStore.MigrationStore (IndexedUserMigrationStore)
import Wire.IndexedUserStore.MigrationStore.ElasticSearch
import Wire.ParseException
import Wire.PostgresMigrationOpts
import Wire.Rpc
import Wire.Sem.Logger.TinyLog
import Wire.Sem.Metrics (Metrics)
import Wire.Sem.Metrics.IO
import Wire.UserKeyStore (UserKeyStore)
import Wire.UserKeyStore.Cassandra
import Wire.UserSearch.Migration (MigrationException)
import Wire.UserStore (UserStore)
import Wire.UserStore.Cassandra
import Wire.UserStore.Postgres (interpretUserStorePostgres)

type BrigIndexEffectStack =
  [ UserKeyStore,
    UserStore,
    IndexedUserStore,
    Error IndexedUserStoreError,
    IndexedUserMigrationStore,
    Error MigrationException,
    GalleyAPIAccess,
    Error ParseException,
    Rpc,
    Metrics,
    TinyLog,
    Input Hasql.Pool,
    Error UsageError,
    Error ClientError,
    Embed IO,
    Final IO
  ]

type SemDeps = (Manager, ClientState, Hasql.Pool, BHEnv, IndexedUserStoreConfig, RequestId, IndexName)

newtype PostgresUsageException = PostgresUsageException UsageError
  deriving (Show)

instance Exception PostgresUsageException

mkSemDeps :: ESConnectionSettings -> CassandraSettings -> PostgresSettings -> Logger -> IO SemDeps
mkSemDeps esConn cas pg logger = do
  mgr <- initHttpManagerWithTLSConfig esConn.esInsecureSkipVerifyTls esConn.esCaCert
  mEsCreds :: Maybe Credentials <- for esConn.esCredentials initCredentials
  casClient <- defInitCassandra (toCassandraOpts cas) logger
  pgPool <- initPostgresPool pg.pool pg.settings pg.passwordFile
  let bhEnv =
        BHEnv
          { bhServer = toESServer esConn.esServer,
            bhManager = mgr,
            bhRequestHook = maybe pure (\creds -> ES.basicAuthHook (ES.EsUsername creds.username) (ES.EsPassword creds.password)) mEsCreds
          }
      indexedUserStoreConfig =
        IndexedUserStoreConfig
          { conn =
              ESConn
                { indexName = esConn.esIndex,
                  env = bhEnv
                },
            additionalConn = Nothing
          }
      reqId = (RequestId "brig-index")
      migrationIndexName = fromMaybe defaultMigrationIndexName (esMigrationIndexName esConn)
  pure (mgr, casClient, pgPool, bhEnv, indexedUserStoreConfig, reqId, migrationIndexName)

runSem :: SemDeps -> UserStorageLocation -> Endpoint -> Logger -> Sem BrigIndexEffectStack a -> IO a
runSem (mgr, casClient, pgPool, bhEnv, indexedUserStoreConfig, reqId, migrationIndexName) userStorage galleyEndpoint logger action = do
  let userStoreInterpreter = case userStorage.userStorageLocation of
        CassandraStorage -> interpretUserStoreCassandra casClient
        MigrationToPostgresql -> error "Migration not implemented for user"
        PostgresqlStorage -> interpretUserStorePostgres
  runFinal
    . embedToFinal
    . throwErrorToIOFinal @ClientError
    . throwPostgresUsageErrorToIOFinal
    . runInputConst pgPool
    . loggerToTinyLogReqId reqId logger
    . ignoreMetrics
    . runRpcWithHttp mgr reqId
    . throwErrorToIOFinal @ParseException
    . interpretGalleyAPIAccessToRpc mempty galleyEndpoint
    . throwErrorToIOFinal @MigrationException
    . interpretIndexedUserMigrationStoreES bhEnv migrationIndexName
    . throwErrorToIOFinal @IndexedUserStoreError
    . interpretIndexedUserStoreES indexedUserStoreConfig
    . userStoreInterpreter
    . interpretUserKeyStoreCassandra casClient
    $ action

throwErrorToIOFinal :: (Exception e, Member (Final IO) r) => InterpreterFor (Error e) r
throwErrorToIOFinal action = do
  runError action >>= \case
    Left e -> embedFinal $ throwIO e
    Right a -> pure a

throwPostgresUsageErrorToIOFinal :: (Member (Final IO) r) => InterpreterFor (Error UsageError) r
throwPostgresUsageErrorToIOFinal action = do
  runError action >>= \case
    Left e -> embedFinal $ throwIO (PostgresUsageException e)
    Right a -> pure a

runCommand :: Logger -> Command -> IO ()
runCommand l = \case
  Create es galley -> do
    e <- initIndex l (es ^. esConnection) galley
    runIndexIO e $ createIndexIfNotPresent (mkCreateIndexSettings es)
  Reset es galley -> do
    e <- initIndex l (es ^. esConnection) galley
    runIndexIO e $ resetIndex (mkCreateIndexSettings es)
  Reindex es cas pg userStorageLocation galley pageSize -> do
    semDeps <- mkSemDeps (es ^. esConnection) cas pg l
    IndexedUserStoreBulk.syncAllUsers (runSem semDeps userStorageLocation galley l) pageSize
  ReindexSameOrNewer es cas pg userStorageLocation galley pageSize -> do
    semDeps <- mkSemDeps (es ^. esConnection) cas pg l
    IndexedUserStoreBulk.forceSyncAllUsers (runSem semDeps userStorageLocation galley l) pageSize
  UpdateMapping esConn galley -> do
    e <- initIndex l esConn galley
    runIndexIO e updateMapping
  Migrate es cas pg userStorageLocation galley pageSize -> do
    semDeps <- mkSemDeps (es ^. esConnection) cas pg l
    IndexedUserStoreBulk.migrateData (runSem semDeps userStorageLocation galley l) pageSize
  ReindexFromAnotherIndex reindexSettings -> do
    mgr <-
      initHttpManagerWithTLSConfig
        (reindexSettings ^. reindexEsConnection . to esInsecureSkipVerifyTls)
        (reindexSettings ^. reindexEsConnection . to esCaCert)
    mCreds <- for (reindexSettings ^. reindexEsConnection . to esCredentials) initCredentials
    let bhEnv = initES (reindexSettings ^. reindexEsConnection . to esServer) mgr mCreds
    ES.runBH bhEnv $ do
      let src = reindexSettings ^. reindexEsConnection . to esIndex
          dest = view reindexDestIndex reindexSettings
          timeoutSeconds = view reindexTimeoutSeconds reindexSettings

      srcExists <- ES.indexExists src
      unless srcExists $ do
        throwM $ ReindexFromAnotherIndexError $ "Source index " <> show src <> " doesn't exist"

      destExists <- ES.indexExists dest
      unless destExists $ do
        throwM $ ReindexFromAnotherIndexError $ "Destination index " <> show dest <> " doesn't exist"

      Log.info l $ Log.msg ("Reindexing" :: ByteString) . Log.field "from" (show src) . Log.field "to" (show dest)
      eitherTaskNodeId <- ES.reindexAsync $ ES.mkReindexRequest src dest
      case eitherTaskNodeId of
        Left e -> throwM $ ReindexFromAnotherIndexError $ "Error occurred while running reindex: " <> show e
        Right taskNodeId -> do
          Log.info l $ Log.field "taskNodeId" (show taskNodeId)
          waitForTaskToComplete @ES.ReindexResponse timeoutSeconds taskNodeId
          Log.info l $ Log.msg ("Finished reindexing" :: ByteString)
  where
    initES esURI mgr mCreds =
      let env = ES.mkBHEnv (toESServer esURI) mgr
       in maybe env (\(creds :: Credentials) -> env {ES.bhRequestHook = ES.basicAuthHook (ES.EsUsername creds.username) (ES.EsPassword creds.password)}) mCreds

initIndex :: Logger -> ESConnectionSettings -> Endpoint -> IO IndexEnv
initIndex l esConn gly = do
  mgr <- initHttpManagerWithTLSConfig esConn.esInsecureSkipVerifyTls esConn.esCaCert
  let esOpts =
        ElasticSearchOpts
          { url = toESServer esConn.esServer,
            index = esConn.esIndex,
            credentials = esConn.esCredentials,
            insecureSkipVerifyTls = esConn.esInsecureSkipVerifyTls,
            caCert = esConn.esCaCert,
            additionalWriteIndex = Nothing,
            additionalWriteIndexUrl = Nothing,
            additionalCredentials = Nothing,
            additionalInsecureSkipVerifyTls = False,
            additionalCaCert = Nothing
          }

  mkIndexEnv esOpts l gly mgr

waitForTaskToComplete :: forall a m. (ES.MonadBH m, MonadThrow m, FromJSON a) => Int -> ES.TaskNodeId -> m ()
waitForTaskToComplete timeoutSeconds taskNodeId = do
  -- Delay is 0.1 seconds, so retries are limited to timeoutSeconds * 10
  let policy = constantDelay 100000 <> limitRetries (timeoutSeconds * 10)
  let retryCondition _ = fmap not . isTaskComplete
  taskEither <- retrying policy retryCondition (const $ ES.getTask @m @a taskNodeId)
  task <- either errTaskGet pure taskEither
  unless (ES.taskResponseCompleted task) $ do
    throwM $ ReindexFromAnotherIndexError $ "Timed out waiting for task: " <> show taskNodeId
  when (isJust $ ES.taskResponseError task) $ do
    throwM $
      ReindexFromAnotherIndexError $
        "Task failed with error: "
          <> UTF8.toString (Aeson.encode $ ES.taskResponseError task)
  where
    isTaskComplete :: Either ES.EsError (ES.TaskResponse a) -> m Bool
    isTaskComplete (Left e) = throwM $ ReindexFromAnotherIndexError $ "Error response while getting task: " <> show e
    isTaskComplete (Right taskRes) = pure $ ES.taskResponseCompleted taskRes

    errTaskGet :: ES.EsError -> m x
    errTaskGet e = throwM $ ReindexFromAnotherIndexError $ "Error response while getting task: " <> show e

newtype ReindexFromAnotherIndexError = ReindexFromAnotherIndexError String
  deriving (Show)

instance Exception ReindexFromAnotherIndexError
