Add support for Mailer and Data Seeder for Lambda deployment

registry-server/src/Registry/Model/Context/ContextLenses.hs

@@ -48,6 +48,9 @@ instance HasField "sentry'" ServerConfig ServerConfigSentry where
 instance HasField "cloud'" ServerConfig ServerConfigCloud where
   getField = (.cloud)
 
+instance HasField "persistentCommand'" ServerConfig ServerConfigPersistentCommand where
+  getField = (.persistentCommand)
+
 instance HasField "dbPool'" AppContext (Pool Connection) where
   getField = (.dbPool)
 

registry-server/src/Registry/Worker/CronWorkers.hs

@@ -6,11 +6,13 @@ import Registry.Model.Context.BaseContext
 import Registry.Model.Context.ContextLenses ()
 import Registry.Service.PersistentCommand.PersistentCommandService
 import Shared.Common.Model.Config.ServerConfig
+import Shared.PersistentCommand.Service.PersistentCommand.PersistentCommandService
 import Shared.Worker.Model.Worker.CronWorker
 
 workers :: [CronWorker BaseContext AppContextM]
 workers =
   [ persistentCommandRetryWorker
+  , persistentCommandRetryLambdaWorker
   ]
 
 -- ------------------------------------------------------------------
@@ -24,3 +26,14 @@ persistentCommandRetryWorker =
     , function = runPersistentCommands'
     , wrapInTransaction = False
     }
+
+persistentCommandRetryLambdaWorker :: CronWorker BaseContext AppContextM
+persistentCommandRetryLambdaWorker =
+  CronWorker
+    { name = "persistentCommandRetryLambdaWorker"
+    , condition = (.serverConfig.persistentCommand.retryLambdaJob.enabled)
+    , cronDefault = "* * * * *"
+    , cron = (.serverConfig.persistentCommand.retryLambdaJob.cron)
+    , function = retryPersistentCommandsForLambda
+    , wrapInTransaction = False
+    }

shared-common/src/Shared/Common/Model/Config/ServerConfig.hs

@@ -53,12 +53,20 @@ data ServerConfigCronWorker = ServerConfigCronWorker
   deriving (Generic, Show)
 
 data ServerConfigPersistentCommand = ServerConfigPersistentCommand
-  { listenerJob :: ServerConfigPersistentCommandListenerJob
+  { lambdaFunctions :: [ServerConfigPersistentCommandLambda]
+  , listenerJob :: ServerConfigPersistentCommandListenerJob
   , retryJob :: ServerConfigCronWorker
+  , retryLambdaJob :: ServerConfigCronWorker
   }
   deriving (Generic, Show)
 
 data ServerConfigPersistentCommandListenerJob = ServerConfigPersistentCommandListenerJob
   { enabled :: Bool
   }
   deriving (Generic, Show)
+
+data ServerConfigPersistentCommandLambda = ServerConfigPersistentCommandLambda
+  { component :: String
+  , functionArn :: String
+  }
+  deriving (Generic, Show)

shared-common/src/Shared/Common/Model/Config/ServerConfigDM.hs

@@ -48,8 +48,10 @@ defaultCloud =
 defaultPersistentCommand :: ServerConfigPersistentCommand
 defaultPersistentCommand =
   ServerConfigPersistentCommand
-    { listenerJob = defaultPersistentCommandListenerJob
+    { lambdaFunctions = []
+    , listenerJob = defaultPersistentCommandListenerJob
     , retryJob = defaultPersistentCommandRetryJob
+    , retryLambdaJob = defaultPersistentCommandRetryLambdaJob
     }
 
 defaultPersistentCommandListenerJob :: ServerConfigPersistentCommandListenerJob
@@ -59,3 +61,7 @@ defaultPersistentCommandListenerJob =
 defaultPersistentCommandRetryJob :: ServerConfigCronWorker
 defaultPersistentCommandRetryJob =
   ServerConfigCronWorker {enabled = True, cron = "* * * * *"}
+
+defaultPersistentCommandRetryLambdaJob :: ServerConfigCronWorker
+defaultPersistentCommandRetryLambdaJob =
+  ServerConfigCronWorker {enabled = True, cron = "* * * * *"}

shared-common/src/Shared/Common/Model/Config/ServerConfigIM.hs

@@ -55,6 +55,8 @@ instance FromEnv ServerConfigPersistentCommand where
       [ \c -> applyEnvVariable "PERSISTENT_COMMAND_LISTENER_JOB_ENABLED" c.listenerJob.enabled (\x -> c {listenerJob = c.listenerJob {enabled = x}} :: ServerConfigPersistentCommand)
       , \c -> applyEnvVariable "PERSISTENT_COMMAND_RETRY_JOB_ENABLED" c.retryJob.enabled (\x -> c {retryJob = c.retryJob {enabled = x}} :: ServerConfigPersistentCommand)
       , \c -> applyStringEnvVariable "PERSISTENT_COMMAND_RETRY_JOB_CRON" c.retryJob.cron (\x -> c {retryJob = c.retryJob {cron = x}} :: ServerConfigPersistentCommand)
+      , \c -> applyEnvVariable "PERSISTENT_COMMAND_RETRY_LAMBDA_JOB_ENABLED" c.retryLambdaJob.enabled (\x -> c {retryLambdaJob = c.retryLambdaJob {enabled = x}} :: ServerConfigPersistentCommand)
+      , \c -> applyStringEnvVariable "PERSISTENT_COMMAND_RETRY_LAMBDA_JOB_CRON" c.retryLambdaJob.cron (\x -> c {retryLambdaJob = c.retryLambdaJob {cron = x}} :: ServerConfigPersistentCommand)
       ]
 
 instance FromEnv ServerConfigLogging where

shared-common/src/Shared/Common/Model/Config/ServerConfigJM.hs

@@ -75,11 +75,20 @@ instance FromJSON ServerConfigCronWorker where
 
 instance FromJSON ServerConfigPersistentCommand where
   parseJSON (Object o) = do
+    lambdaFunctions <- o .:? "lambdaFunctions" .!= []
     listenerJob <- o .:? "listenerJob" .!= defaultPersistentCommand.listenerJob
     retryJob <- o .:? "retryJob" .!= defaultPersistentCommand.retryJob
+    retryLambdaJob <- o .:? "retryLambdaJob" .!= defaultPersistentCommand.retryLambdaJob
     return ServerConfigPersistentCommand {..}
   parseJSON _ = mzero
 
+instance FromJSON ServerConfigPersistentCommandLambda where
+  parseJSON (Object o) = do
+    component <- o .: "component"
+    functionArn <- o .: "functionArn"
+    return ServerConfigPersistentCommandLambda {..}
+  parseJSON _ = mzero
+
 instance FromJSON ServerConfigPersistentCommandListenerJob where
   parseJSON (Object o) = do
     enabled <- o .:? "enabled" .!= defaultPersistentCommandListenerJob.enabled

shared-common/src/Shared/Common/Model/Context/AppContext.hs

@@ -26,6 +26,7 @@ class
   , HasField "serverConfig'" context sc
   , HasField "cloud'" sc ServerConfigCloud
   , HasField "s3'" sc ServerConfigS3
+  , HasField "persistentCommand'" sc ServerConfigPersistentCommand
   , HasField "sentry'" sc ServerConfigSentry
   , HasField "buildInfoConfig'" context BuildInfoConfig
   , HasField "httpClientManager'" context Manager

shared-common/src/Shared/PersistentCommand/Database/DAO/PersistentCommand/PersistentCommandDAO.hs

@@ -1,6 +1,8 @@
 module Shared.PersistentCommand.Database.DAO.PersistentCommand.PersistentCommandDAO where
 
+import Control.Monad.Reader (ask)
 import qualified Data.ByteString.Char8 as BS
+import qualified Data.List as L
 import Data.String (fromString)
 import qualified Data.UUID as U
 import Database.PostgreSQL.Simple
@@ -14,13 +16,17 @@ import Shared.Common.Database.DAO.Common
 import Shared.Common.Model.Common.Page
 import Shared.Common.Model.Common.Pageable
 import Shared.Common.Model.Common.Sort
+import Shared.Common.Model.Config.ServerConfig
 import Shared.Common.Model.Context.AppContext
 import Shared.Common.Util.Logger
-import Shared.Common.Util.String (trim)
+import Shared.Common.Util.String (f'', trim)
+import Shared.PersistentCommand.Database.Mapping.PersistentCommand.LambdaInvocationResult ()
 import Shared.PersistentCommand.Database.Mapping.PersistentCommand.PersistentCommand ()
 import Shared.PersistentCommand.Database.Mapping.PersistentCommand.PersistentCommandSimple ()
+import Shared.PersistentCommand.Model.PersistentCommand.LambdaInvocationResult
 import Shared.PersistentCommand.Model.PersistentCommand.PersistentCommand
 import Shared.PersistentCommand.Model.PersistentCommand.PersistentCommandSimple
+import Shared.PersistentCommand.Service.PersistentCommand.PersistentCommandMapper
 
 entityName = "persistent_command"
 
@@ -40,19 +46,35 @@ findPersistentCommandsPage states pageable sort = do
           _ -> f' "WHERE state in (%s) " [generateQuestionMarks states]
   createFindEntitiesPageableQuerySortFn entityName pageLabel pageable sort "*" condition states
 
-findPersistentCommandsByStates :: (AppContextC s sc m, FromField identity) => m [PersistentCommandSimple identity]
-findPersistentCommandsByStates = do
+findPersistentCommandsForRetryByStates :: (AppContextC s sc m, FromField identity) => m [PersistentCommandSimple identity]
+findPersistentCommandsForRetryByStates = findPersistentCommandsByStates True []
+
+findPersistentCommandsForLambdaByStates :: (AppContextC s sc m, FromField identity) => [String] -> m [PersistentCommandSimple identity]
+findPersistentCommandsForLambdaByStates = findPersistentCommandsByStates False
+
+findPersistentCommandsByStates :: (AppContextC s sc m, FromField identity) => Bool -> [String] -> m [PersistentCommandSimple identity]
+findPersistentCommandsByStates internal components = do
+  let componentCondition =
+        case components of
+          [] -> ""
+          _ -> f' "AND component IN (%s) " [generateQuestionMarks components]
   let sql =
-        "SELECT uuid, destination, tenant_uuid, created_by \
-        \FROM persistent_command \
-        \WHERE (state = 'NewPersistentCommandState' \
-        \  OR (state = 'ErrorPersistentCommandState' AND attempts < max_attempts AND updated_at < (now() - (2 ^ attempts - 1) * INTERVAL '1 min'))) \
-        \  AND internal = true \
-        \ORDER BY created_at \
-        \LIMIT 5 \
-        \FOR UPDATE"
-  logInfoI _CMP_DATABASE (trim sql)
-  let action conn = query_ conn (fromString sql)
+        fromString $
+          f''
+            "SELECT uuid, destination, component, tenant_uuid, created_by \
+            \FROM persistent_command \
+            \WHERE (state = 'NewPersistentCommandState' \
+            \  OR (state = 'ErrorPersistentCommandState' AND attempts < max_attempts AND updated_at < (now() - (2 ^ attempts - 1) * INTERVAL '1 min'))) \
+            \  AND internal = ${internal} ${componentCondition} \
+            \ORDER BY created_at \
+            \LIMIT 5 \
+            \FOR UPDATE"
+            [ ("internal", show internal)
+            , ("componentCondition", componentCondition)
+            ]
+  let params = components
+  logQuery sql params
+  let action conn = query conn sql params
   runDB action
 
 findPersistentCommandByUuid :: (AppContextC s sc m, FromField identity) => U.UUID -> m (PersistentCommand identity)
@@ -68,9 +90,11 @@ findPersistentCommandSimpleByUuid uuid =
 insertPersistentCommand :: (AppContextC s sc m, ToField identity) => PersistentCommand identity -> m Int64
 insertPersistentCommand command = do
   createInsertFn entityName command
-  if command.internal
-    then notifyPersistentCommandQueue
-    else notifySpecificPersistentCommandQueue command
+  context <- ask
+  case (command.internal, L.find (\lf -> lf.component == command.component) (context.serverConfig'.persistentCommand'.lambdaFunctions)) of
+    (True, _) -> notifyPersistentCommandQueue
+    (False, Nothing) -> notifySpecificPersistentCommandQueue command
+    (False, Just lf) -> invokeLambdaFunction (toSimple command) lf
 
 updatePersistentCommandByUuid :: (AppContextC s sc m, ToField identity) => PersistentCommand identity -> m Int64
 updatePersistentCommandByUuid command = do
@@ -125,3 +149,12 @@ notifySpecificPersistentCommandQueue command = do
   logInfoI _CMP_DATABASE (trim sql)
   let action conn = execute_ conn (fromString sql)
   runDB action
+
+invokeLambdaFunction :: AppContextC s sc m => PersistentCommandSimple identity -> ServerConfigPersistentCommandLambda -> m Int64
+invokeLambdaFunction command lf = do
+  let sql = f' "SELECT * FROM aws_lambda.invoke('%s', '{}'::json, 'eu-central-1', 'Event');" [lf.functionArn]
+  logInfoI _CMP_DATABASE (trim sql)
+  let action conn = query_ conn (fromString sql)
+  result <- runDB action :: AppContextC s sc m => m [LambdaInvocationResult]
+  logInfoI _CMP_DATABASE (f' "Lambda function '%s' invoked with result: '%s'" [lf.functionArn, show result])
+  return 1

shared-common/src/Shared/PersistentCommand/Database/Mapping/PersistentCommand/LambdaInvocationResult.hs

@@ -0,0 +1,7 @@
+module Shared.PersistentCommand.Database.Mapping.PersistentCommand.LambdaInvocationResult where
+
+import Database.PostgreSQL.Simple
+
+import Shared.PersistentCommand.Model.PersistentCommand.LambdaInvocationResult
+
+instance FromRow LambdaInvocationResult

shared-common/src/Shared/PersistentCommand/Model/PersistentCommand/LambdaInvocationResult.hs

@@ -0,0 +1,12 @@
+module Shared.PersistentCommand.Model.PersistentCommand.LambdaInvocationResult where
+
+import GHC.Generics
+import Data.Aeson
+
+data LambdaInvocationResult = LambdaInvocationResult
+  { statusCode :: Int
+  , payload :: Maybe Value
+  , executedVersion :: Maybe String
+  , logResult :: Maybe String
+  }
+  deriving (Show, Generic)

shared-common/src/Shared/PersistentCommand/Model/PersistentCommand/PersistentCommandSimple.hs

@@ -6,6 +6,7 @@ import GHC.Generics
 data PersistentCommandSimple identity = PersistentCommandSimple
   { uuid :: U.UUID
   , destination :: Maybe String
+  , component :: String
   , tenantUuid :: U.UUID
   , createdBy :: Maybe identity
   }

shared-common/src/Shared/PersistentCommand/Service/PersistentCommand/PersistentCommandMapper.hs

@@ -43,6 +43,7 @@ toSimple command =
   PersistentCommandSimple
     { uuid = command.uuid
     , destination = command.destination
+    , component = command.component
     , tenantUuid = command.tenantUuid
     , createdBy = command.createdBy
     }

shared-common/src/Shared/PersistentCommand/Service/PersistentCommand/PersistentCommandService.hs

@@ -1,11 +1,12 @@
 module Shared.PersistentCommand.Service.PersistentCommand.PersistentCommandService where
 
 import qualified Control.Exception.Base as E
-import Control.Monad (forever, unless, when)
+import Control.Monad (forever, unless, void, when)
 import Control.Monad.Reader (ask, liftIO)
 import Data.Aeson (Value (..), toJSON)
 import Data.Foldable (traverse_)
 import qualified Data.HashMap.Strict as HashMap
+import qualified Data.List as L
 import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Data.Time
@@ -34,7 +35,7 @@ runPersistentCommands
   -> m ()
 runPersistentCommands runAppContextWithAppContext' updateContext createPersistentCommand execute = do
   checkPermission _DEV_PERM
-  commands <- findPersistentCommandsByStates
+  commands <- findPersistentCommandsForRetryByStates
   unless
     (null commands)
     ( do
@@ -142,6 +143,20 @@ runPersistentCommandChannelListener runAppContextWithAppContext updateContext cr
     _ <- getChannelNotification
     runPersistentCommands runAppContextWithAppContext updateContext createPersistentCommand execute
 
+retryPersistentCommandsForLambda :: AppContextC s sc m => m ()
+retryPersistentCommandsForLambda = do
+  context <- ask
+  let components = fmap (\lf -> lf.component) context.serverConfig'.persistentCommand'.lambdaFunctions
+  persistentCommands <- findPersistentCommandsForLambdaByStates components
+  traverse_ retryPersistentCommandForLambda persistentCommands
+
+retryPersistentCommandForLambda :: AppContextC s sc m => PersistentCommandSimple U.UUID -> m ()
+retryPersistentCommandForLambda command = do
+  context <- ask
+  case L.find (\lf -> lf.component == command.component) (context.serverConfig'.persistentCommand'.lambdaFunctions) of
+    Just lf -> void $ invokeLambdaFunction command lf
+    Nothing -> logWarnI _CMP_DATABASE (f' "No lambda function found for persistent command '%s'" [U.toString command.uuid])
+
 -- --------------------------------
 -- PRIVATE
 -- --------------------------------

wizard-server/src/Wizard/Model/Context/ContextLenses.hs

@@ -48,6 +48,9 @@ instance HasField "sentry'" ServerConfig ServerConfigSentry where
 instance HasField "cloud'" ServerConfig ServerConfigCloud where
   getField = (.cloud)
 
+instance HasField "persistentCommand'" ServerConfig ServerConfigPersistentCommand where
+  getField = (.persistentCommand)
+
 instance HasField "dbPool'" AppContext (Pool Connection) where
   getField = (.dbPool)
 

wizard-server/src/Wizard/Worker/CronWorkers.hs

@@ -1,6 +1,7 @@
 module Wizard.Worker.CronWorkers where
 
 import Shared.Common.Model.Config.ServerConfig
+import Shared.PersistentCommand.Service.PersistentCommand.PersistentCommandService
 import Shared.Worker.Model.Worker.CronWorker
 import Wizard.Cache.CacheUtil
 import Wizard.Model.Cache.ServerCache
@@ -29,6 +30,7 @@ workers =
   , documentWorker
   , feedbackWorker
   , persistentCommandRetryWorker
+  , persistentCommandRetryLambdaWorker
   , cleanQuestionnaireWorker
   , recomputeQuestionnaireIndicationWorker
   , squashQuestionnaireEventsWorker
@@ -106,6 +108,17 @@ persistentCommandRetryWorker =
     , wrapInTransaction = False
     }
 
+persistentCommandRetryLambdaWorker :: CronWorker BaseContext AppContextM
+persistentCommandRetryLambdaWorker =
+  CronWorker
+    { name = "persistentCommandRetryLambdaWorker"
+    , condition = (.serverConfig.persistentCommand.retryLambdaJob.enabled)
+    , cronDefault = "* * * * *"
+    , cron = (.serverConfig.persistentCommand.retryLambdaJob.cron)
+    , function = retryPersistentCommandsForLambda
+    , wrapInTransaction = False
+    }
+
 cleanQuestionnaireWorker :: CronWorker BaseContext AppContextM
 cleanQuestionnaireWorker =
   CronWorker