mirror of
https://github.com/hasura/graphql-engine.git
synced 2024-12-17 12:31:52 +03:00
214 lines
7.7 KiB
Haskell
214 lines
7.7 KiB
Haskell
module Hasura.RQL.DDL.EventTrigger
|
|
( CreateEventTriggerQuery
|
|
, runCreateEventTriggerQuery
|
|
, DeleteEventTriggerQuery
|
|
, runDeleteEventTriggerQuery
|
|
, dropEventTriggerInMetadata
|
|
, RedeliverEventQuery
|
|
, runRedeliverEvent
|
|
, runInvokeEventTrigger
|
|
|
|
-- TODO(from master): review
|
|
, archiveEvents
|
|
, getEventTriggerDef
|
|
) where
|
|
|
|
import Hasura.Prelude
|
|
|
|
import qualified Data.HashMap.Strict as HM
|
|
import qualified Data.HashMap.Strict.InsOrd as OMap
|
|
import qualified Database.PG.Query as Q
|
|
|
|
import Control.Lens ((.~))
|
|
import Data.Aeson
|
|
import Data.Text.Extended
|
|
|
|
|
|
import Hasura.Backends.Postgres.DDL.Table
|
|
import Hasura.Backends.Postgres.Execute.Types
|
|
import Hasura.Backends.Postgres.SQL.Types
|
|
import Hasura.EncJSON
|
|
import Hasura.RQL.Types
|
|
import Hasura.Session
|
|
|
|
archiveEvents :: TriggerName -> Q.TxE QErr ()
|
|
archiveEvents trn =
|
|
Q.unitQE defaultTxErrorHandler [Q.sql|
|
|
UPDATE hdb_catalog.event_log
|
|
SET archived = 't'
|
|
WHERE trigger_name = $1
|
|
|] (Identity trn) False
|
|
|
|
checkEvent :: EventId -> Q.TxE QErr ()
|
|
checkEvent eid = do
|
|
events <- Q.listQE defaultTxErrorHandler
|
|
[Q.sql|
|
|
SELECT l.locked IS NOT NULL AND l.locked >= (NOW() - interval '30 minute')
|
|
FROM hdb_catalog.event_log l
|
|
WHERE l.id = $1
|
|
|] (Identity eid) True
|
|
event <- getEvent events
|
|
assertEventUnlocked event
|
|
where
|
|
getEvent [] = throw400 NotExists "event not found"
|
|
getEvent (x:_) = return x
|
|
|
|
assertEventUnlocked (Identity locked) = when locked $
|
|
throw400 Busy "event is already being processed"
|
|
|
|
markForDelivery :: EventId -> Q.TxE QErr ()
|
|
markForDelivery eid =
|
|
Q.unitQE defaultTxErrorHandler [Q.sql|
|
|
UPDATE hdb_catalog.event_log
|
|
SET
|
|
delivered = 'f',
|
|
error = 'f',
|
|
tries = 0
|
|
WHERE id = $1
|
|
|] (Identity eid) True
|
|
|
|
resolveEventTriggerQuery :: (UserInfoM m, QErrM m, CacheRM m) => CreateEventTriggerQuery -> m (TableCoreInfo 'Postgres, Bool, EventTriggerConf)
|
|
resolveEventTriggerQuery (CreateEventTriggerQuery source name qt insert update delete enableManual retryConf webhook webhookFromEnv mheaders replace) = do
|
|
ti <- askTableCoreInfo source qt
|
|
-- can only replace for same table
|
|
when replace $ do
|
|
ti' <- _tiCoreInfo <$> askTabInfoFromTrigger source name
|
|
when (_tciName ti' /= _tciName ti) $ throw400 NotSupported "cannot replace table or schema for trigger"
|
|
|
|
assertCols ti insert
|
|
assertCols ti update
|
|
assertCols ti delete
|
|
|
|
let rconf = fromMaybe defaultRetryConf retryConf
|
|
return (ti, replace, EventTriggerConf name (TriggerOpsDef insert update delete enableManual) webhook webhookFromEnv rconf mheaders)
|
|
where
|
|
assertCols ti opSpec = onJust opSpec \sos -> case sosColumns sos of
|
|
SubCStar -> return ()
|
|
SubCArray pgcols -> forM_ pgcols (assertPGCol (_tciFieldInfoMap ti) "")
|
|
|
|
createEventTriggerQueryMetadata
|
|
:: (QErrM m, UserInfoM m, CacheRWM m, MetadataM m)
|
|
=> CreateEventTriggerQuery -> m (TableCoreInfo 'Postgres, EventTriggerConf)
|
|
createEventTriggerQueryMetadata q = do
|
|
(tableCoreInfo, replace, triggerConf) <- resolveEventTriggerQuery q
|
|
let table = cetqTable q
|
|
source = cetqSource q
|
|
triggerName = etcName triggerConf
|
|
metadataObj = MOSourceObjId source $ SMOTableObj table $ MTOTrigger triggerName
|
|
buildSchemaCacheFor metadataObj
|
|
$ MetadataModifier
|
|
$ tableMetadataSetter source table.tmEventTriggers %~
|
|
if replace then ix triggerName .~ triggerConf
|
|
else OMap.insert triggerName triggerConf
|
|
pure (tableCoreInfo, triggerConf)
|
|
|
|
runCreateEventTriggerQuery
|
|
:: (QErrM m, UserInfoM m, CacheRWM m, MetadataM m)
|
|
=> CreateEventTriggerQuery -> m EncJSON
|
|
runCreateEventTriggerQuery q = do
|
|
void $ createEventTriggerQueryMetadata q
|
|
pure successMsg
|
|
|
|
runDeleteEventTriggerQuery
|
|
:: (MonadError QErr m, CacheRWM m, MonadIO m, MetadataM m)
|
|
=> DeleteEventTriggerQuery -> m EncJSON
|
|
runDeleteEventTriggerQuery (DeleteEventTriggerQuery source name) = do
|
|
-- liftTx $ delEventTriggerFromCatalog name
|
|
sourceInfo <- askSourceInfo source
|
|
let maybeTable = HM.lookup name $ HM.unions $
|
|
flip map (HM.toList $ _siTables sourceInfo) $ \(table, tableInfo) ->
|
|
HM.map (const table) $ _tiEventTriggerInfoMap tableInfo
|
|
table <- onNothing maybeTable $ throw400 NotExists $
|
|
"event trigger with name " <> name <<> " not exists"
|
|
|
|
withNewInconsistentObjsCheck
|
|
$ buildSchemaCache
|
|
$ MetadataModifier
|
|
$ tableMetadataSetter source table %~ dropEventTriggerInMetadata name
|
|
|
|
liftEitherM $ liftIO $ runPgSourceWriteTx (_siConfiguration sourceInfo) $ do
|
|
delTriggerQ name
|
|
archiveEvents name
|
|
pure successMsg
|
|
|
|
dropEventTriggerInMetadata :: TriggerName -> TableMetadata b -> TableMetadata b
|
|
dropEventTriggerInMetadata name =
|
|
tmEventTriggers %~ OMap.delete name
|
|
|
|
deliverEvent ::EventId -> Q.TxE QErr ()
|
|
deliverEvent eventId = do
|
|
checkEvent eventId
|
|
markForDelivery eventId
|
|
|
|
runRedeliverEvent
|
|
:: (MonadIO m, CacheRM m, QErrM m, MetadataM m)
|
|
=> RedeliverEventQuery -> m EncJSON
|
|
runRedeliverEvent (RedeliverEventQuery eventId source) = do
|
|
sourceConfig <- askSourceConfig source
|
|
liftEitherM $ liftIO $ runPgSourceWriteTx sourceConfig $ deliverEvent eventId
|
|
pure successMsg
|
|
|
|
insertManualEvent
|
|
:: QualifiedTable
|
|
-> TriggerName
|
|
-> Value
|
|
-> Q.TxE QErr EventId
|
|
insertManualEvent qt trn rowData = do
|
|
let op = tshow MANUAL
|
|
eids <- map runIdentity <$> Q.listQE defaultTxErrorHandler [Q.sql|
|
|
SELECT hdb_catalog.insert_event_log($1, $2, $3, $4, $5)
|
|
|] (sn, tn, trn, op, Q.AltJ $ toJSON rowData) True
|
|
getEid eids
|
|
where
|
|
QualifiedObject sn tn = qt
|
|
getEid [] = throw500 "could not create manual event"
|
|
getEid (x:_) = return x
|
|
|
|
runInvokeEventTrigger
|
|
:: (MonadIO m, QErrM m, CacheRM m, MetadataM m)
|
|
=> InvokeEventTriggerQuery -> m EncJSON
|
|
runInvokeEventTrigger (InvokeEventTriggerQuery name source payload) = do
|
|
trigInfo <- askEventTriggerInfo source name
|
|
assertManual $ etiOpsDef trigInfo
|
|
ti <- askTabInfoFromTrigger source name
|
|
sourceConfig <- askSourceConfig source
|
|
eid <- liftEitherM $ liftIO $ runPgSourceWriteTx sourceConfig $
|
|
insertManualEvent (_tciName $ _tiCoreInfo ti) name payload
|
|
return $ encJFromJValue $ object ["event_id" .= eid]
|
|
where
|
|
assertManual (TriggerOpsDef _ _ _ man) = case man of
|
|
Just True -> return ()
|
|
_ -> throw400 NotSupported "manual mode is not enabled for event trigger"
|
|
|
|
getEventTriggerDef
|
|
:: TriggerName
|
|
-> Q.TxE QErr (QualifiedTable, EventTriggerConf)
|
|
getEventTriggerDef triggerName = do
|
|
(sn, tn, Q.AltJ etc) <- Q.getRow <$> Q.withQE defaultTxErrorHandler
|
|
[Q.sql|
|
|
SELECT e.schema_name, e.table_name, e.configuration::json
|
|
FROM hdb_catalog.event_triggers e where e.name = $1
|
|
|] (Identity triggerName) False
|
|
return (QualifiedObject sn tn, etc)
|
|
|
|
askTabInfoFromTrigger
|
|
:: (QErrM m, CacheRM m)
|
|
=> SourceName -> TriggerName -> m (TableInfo 'Postgres)
|
|
askTabInfoFromTrigger sourceName trn = do
|
|
sc <- askSchemaCache
|
|
let tabInfos = HM.elems $ fromMaybe mempty $ unsafeTableCache sourceName $ scSources sc
|
|
find (isJust . HM.lookup trn . _tiEventTriggerInfoMap) tabInfos
|
|
`onNothing` throw400 NotExists errMsg
|
|
where
|
|
errMsg = "event trigger " <> triggerNameToTxt trn <<> " does not exist"
|
|
|
|
askEventTriggerInfo
|
|
:: (QErrM m, CacheRM m)
|
|
=> SourceName -> TriggerName -> m (EventTriggerInfo 'Postgres)
|
|
askEventTriggerInfo sourceName trn = do
|
|
ti <- askTabInfoFromTrigger sourceName trn
|
|
let etim = _tiEventTriggerInfoMap ti
|
|
HM.lookup trn etim `onNothing` throw400 NotExists errMsg
|
|
where
|
|
errMsg = "event trigger " <> triggerNameToTxt trn <<> " does not exist"
|