Repository: hasura/graphql-engine
File: server/src-lib/Hasura/Eventing/ScheduledTrigger.hs (Haskell)

Code before:
                                       webhook
                                       payload'
                                       ctiRetryConf
                                       ctiHeaders
                                       ctiComment
            finally <- runExceptT $
              runReaderT (processScheduledEvent logEnv pgpool scheduledEvent CronScheduledEvent) (logger, httpMgr)
            removeEventFromLockedEvents id'
            either logInternalError pure finally
    Left err -> logInternalError err
  where
    logInternalError err = L.unLogger logger $ ScheduledTriggerInternalErr err

    saveLockedCronEvents :: [CronEventPartial] -> IO ()
    saveLockedCronEvents partialCronEvents =
      liftIO $ atomically $ do
        lockedEvents <- readTVar lockedCronEvents
        let ids = map cepId partialCronEvents
        writeTVar lockedCronEvents $!
          Set.union lockedEvents $ Set.fromList ids

    removeEventFromLockedEvents :: CronEventId -> IO ()
    removeEventFromLockedEvents eventId =
      liftIO $ atomically $ do
        lockedEvents <- readTVar lockedCronEvents
        writeTVar lockedCronEvents $! Set.delete eventId lockedEvents

processStandAloneEvents
  :: HasVersion
  => L.Logger L.Hasura
  -> LogEnvHeaders
  -> HTTP.Manager
  -> Q.PGPool
  -> TVar (Set.Set StandAloneScheduledEventId)
  -> IO ()
processStandAloneEvents logger logEnv httpMgr pgpool lockedStandAloneEvents = do
  standAloneScheduledEvents <-
    runExceptT $
      Q.runTx pgpool (Q.ReadCommitted, Just Q.ReadWrite) getOneOffScheduledEvents
  case standAloneScheduledEvents of
    Right standAloneScheduledEvents' -> do
      -- save the locked standalone events that have been fetched from the
      -- database, the events stored here will be unlocked in case a
      -- graceful shutdown is initiated in midst of processing these events
      saveLockedStandAloneEvents standAloneScheduledEvents'
      for_ standAloneScheduledEvents' $
             \(StandAloneScheduledEvent id'
                                        scheduledTime
                                        tries
                                        webhookConf
                                        payload

Review comment (refactor):
There’s quite a lot of duplicate logic between `processEventQueue`, `processCronEvents`, and `processStandAloneEvents`. Would it be possible to share more of that logic? It seems like maybe there’s a potential “event processor” abstraction here that could simplify a lot of this code!

Code after:
  -> LogEnvHeaders
  -> HTTP.Manager
  -> Q.PGPool
  -> TVar (Set.Set StandAloneScheduledEventId)
  -> IO ()
processStandAloneEvents logger logEnv httpMgr pgpool lockedStandAloneEvents = do
  standAloneScheduledEvents <-
    runExceptT $
      Q.runTx pgpool (Q.ReadCommitted, Just Q.ReadWrite) getOneOffScheduledEvents
  case standAloneScheduledEvents of
    Right standAloneScheduledEvents' -> do
      -- save the locked standalone events that have been fetched from the
      -- database, the events stored here will be unlocked in case a
      -- graceful shutdown is initiated in midst of processing these events
      saveLockedEvents (map saseId standAloneScheduledEvents') lockedStandAloneEvents
      for_ standAloneScheduledEvents' $
             \(StandAloneScheduledEvent id'
                                        scheduledTime
                                        tries
                                        webhookConf
                                        payload
                                        retryConf
                                        headerConf
                                        comment  )
        -> do
        webhookInfo <- runExceptT $ resolveWebhook webhookConf
        headerInfo <- runExceptT $ getHeaderInfosFromConf headerConf

        case webhookInfo of
          Right webhookInfo' -> do
            case headerInfo of
              Right headerInfo' -> do
                let webhook = unResolvedWebhook webhookInfo'
                    payload' = fromMaybe J.Null payload
                    scheduledEvent = ScheduledEventFull id'
                                                        Nothing
                                                        scheduledTime
                                                        tries
                                                        webhook
                                                        payload'
                                                        retryConf
                                                        headerInfo'
                                                        comment
                finally <- runExceptT $
                  runReaderT (processScheduledEvent logEnv pgpool scheduledEvent StandAloneEvent) $
                                 (logger, httpMgr)
                removeEventFromLockedEvents id' lockedStandAloneEvents
                either logInternalError pure finally

              Left headerInfoErr -> logInternalError headerInfoErr