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

Code before:
                   "Events processor may not be keeping up with events generated in postgres, " <>
                   "or we're working on a backlog of events. Consider increasing " <>
                   "HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE"
             go eventsNext (fullFetchCount+1) (alreadyWarned || clearlyBehind)

         | otherwise -> do
             when (lenEvents /= fetchBatchSize && alreadyWarned) $
               -- emit as warning in case users are only logging warning severity and saw above
               L.unLogger logger $ L.UnstructuredLog L.LevelWarn $ fromString $
                 "It looks like the events processor is keeping up again."
             go eventsNext 0 False

    processEvent
      :: forall io r
       . ( HasVersion
         , MonadIO io
         , MonadReader r io
         , Has HTTP.Manager r
         , Has (L.Logger L.Hasura) r
         , Tracing.MonadTrace io
         )
      => Event -> io ()
    processEvent e = do
      cache <- liftIO getSchemaCache
      let meti = getEventTriggerInfoFromEvent cache e
      case meti of
        Nothing -> do
          --  This rare error can happen in the following known cases:
          --  i) schema cache is not up-to-date (due to some bug, say during schema syncing across multiple instances)
          --  ii) the event trigger is dropped when this event was just fetched
          logQErr $ err500 Unexpected "table or event-trigger not found in schema cache"
          liftIO . runExceptT $ Q.runTx pool (Q.RepeatableRead, Just Q.ReadWrite) $ do
            currentTime <- liftIO getCurrentTime
            -- For such an event, we unlock the event and retry after a minute
            setRetry e (addUTCTime 60 currentTime)
          >>= flip onLeft logQErr
        Just eti -> do
          let webhook = T.unpack $ wciCachedValue $ etiWebhookInfo eti
              retryConf = etiRetryConf eti
              timeoutSeconds = fromMaybe defaultTimeoutSeconds (rcTimeoutSec retryConf)
              responseTimeout = HTTP.responseTimeoutMicro (timeoutSeconds * 1000000)
              headerInfos = etiHeaders eti
              etHeaders = map encodeHeader headerInfos
              headers = addDefaultHeaders etHeaders
              ep = createEventPayload retryConf e
              payload = encode $ toJSON ep
              extraLogCtx = ExtraLogContext Nothing (epId ep) -- avoiding getting current time here to avoid another IO call with each event call
              requestDetails = RequestDetails $ LBS.length payload
          res <- runExceptT $ tryWebhook headers responseTimeout payload webhook
          logHTTPForET res extraLogCtx requestDetails
          let decodedHeaders = map (decodeHeader logenv headerInfos) headers

Review comment (performance):
```suggestion
      case getEventTriggerInfoFromEvent cache e of
```

Code after:
                   "Events processor may not be keeping up with events generated in postgres, " <>
                   "or we're working on a backlog of events. Consider increasing " <>
                   "HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE"
             go eventsNext (fullFetchCount+1) (alreadyWarned || clearlyBehind)

         | otherwise -> do
             when (lenEvents /= fetchBatchSize && alreadyWarned) $
               -- emit as warning in case users are only logging warning severity and saw above
               L.unLogger logger $ L.UnstructuredLog L.LevelWarn $ fromString $
                 "It looks like the events processor is keeping up again."
             go eventsNext 0 False

    processEvent
      :: forall io r
       . ( HasVersion
         , MonadIO io
         , MonadReader r io
         , Has HTTP.Manager r
         , Has (L.Logger L.Hasura) r
         , Tracing.MonadTrace io
         )
      => Event -> io ()
    processEvent e = do
      cache <- liftIO getSchemaCache
      let meti = getEventTriggerInfoFromEvent cache e
      case meti of
        Left err -> do
          --  This rare error can happen in the following known cases:
          --  i) schema cache is not up-to-date (due to some bug, say during schema syncing across multiple instances)
          --  ii) the event trigger is dropped when this event was just fetched
          logQErr $ err500 Unexpected err
          liftIO . runExceptT $ Q.runTx pool (Q.RepeatableRead, Just Q.ReadWrite) $ do
            currentTime <- liftIO getCurrentTime
            -- For such an event, we unlock the event and retry after a minute
            setRetry e (addUTCTime 60 currentTime)
          >>= flip onLeft logQErr
        Right eti -> do
          let webhook = T.unpack $ wciCachedValue $ etiWebhookInfo eti
              retryConf = etiRetryConf eti
              timeoutSeconds = fromMaybe defaultTimeoutSeconds (rcTimeoutSec retryConf)
              responseTimeout = HTTP.responseTimeoutMicro (timeoutSeconds * 1000000)
              headerInfos = etiHeaders eti
              etHeaders = map encodeHeader headerInfos
              headers = addDefaultHeaders etHeaders
              ep = createEventPayload retryConf e
              payload = encode $ toJSON ep
              extraLogCtx = ExtraLogContext Nothing (epId ep) -- avoiding getting current time here to avoid another IO call with each event call
              requestDetails = RequestDetails $ LBS.length payload
          res <- runExceptT $ tryWebhook headers responseTimeout payload webhook
          logHTTPForET res extraLogCtx requestDetails
          let decodedHeaders = map (decodeHeader logenv headerInfos) headers