updated HasResolveCustomSource PR for latest branch

server/src-lib/Hasura/App.hs

@@ -61,6 +61,7 @@ import           Hasura.GraphQL.Transport.HTTP.Protocol    (toParsed)
 import           Hasura.Logging
 import           Hasura.Prelude
 import           Hasura.RQL.DDL.Schema.Cache
+import           Hasura.RQL.DDL.Schema.Source
 import           Hasura.RQL.Types
 import           Hasura.RQL.Types.Run
 import           Hasura.Server.API.Query
@@ -228,7 +229,7 @@ initialiseCtx env (HGEOptionsG rci metadataDbUrl hgeCmd) = do
     migrateCatalogSchema logger pgLogger env connParams rci metadataDbUrl
 
   schemaCacheE <- runExceptT
-    $ peelMetadataRun (RunCtx adminUserInfo httpManager sqlGenCtx) metadata
+    $ peelMetadataRun (RunCtx adminUserInfo httpManager sqlGenCtx {- TODO: -} defaultResolveCustomSource) metadata
     $ buildRebuildableSchemaCache env
 
   schemaCache <- fmap fst $ onLeft schemaCacheE $ \err -> do
@@ -326,6 +327,7 @@ runHGEServer
      , Tracing.HasReporter m
      , MonadQueryInstrumentation m
      , MonadMetadataStorage m
+     , HasResolveCustomSource m
      )
   => Env.Environment
   -> ServeOptions impl
@@ -357,6 +359,8 @@ runHGEServer env ServeOptions{..} InitCtx{..} _ initTime
       Loggers loggerCtx logger _ = _icLoggers
       SchemaSyncCtx schemaSyncListenerThread schemaSyncEventRef = schemaSyncCtx
 
+  srcResolver <- askResolveCustomSource
+
   authModeRes <- runExceptT $ setupAuthMode soAdminSecret soAuthHook soJwtSecret soUnAuthRole
                               _icHttpManager logger
 
@@ -393,7 +397,7 @@ runHGEServer env ServeOptions{..} InitCtx{..} _ initTime
 
   -- start background thread for schema sync event processing
   schemaSyncProcessorThread <-
-    startSchemaSyncProcessorThread sqlGenCtx
+    startSchemaSyncProcessorThread sqlGenCtx srcResolver
     logger _icHttpManager schemaSyncEventRef cacheRef _icInstanceId
 
   let
@@ -631,7 +635,7 @@ runAsAdmin
   -> m (Either QErr a)
 runAsAdmin sqlGenCtx httpManager metadata m =
   fmap (fmap fst) $ runExceptT $
-    peelMetadataRun (RunCtx adminUserInfo httpManager sqlGenCtx) metadata m
+    peelMetadataRun (RunCtx adminUserInfo httpManager sqlGenCtx {- TODO: -} defaultResolveCustomSource) metadata m
 
 execQuery
   :: ( HasVersion
@@ -646,7 +650,7 @@ execQuery env httpManager metadata queryBs = runExceptT do
   QueryWithSource source query <- case A.decode queryBs of
     Just jVal -> decodeValue jVal
     Nothing   -> throw400 InvalidJSON "invalid json"
-  let runCtx = RunCtx adminUserInfo httpManager (SQLGenCtx False)
+  let runCtx = RunCtx adminUserInfo httpManager (SQLGenCtx False) {- TODO: -} defaultResolveCustomSource
       actionM = do
         buildSchemaCacheStrict noMetadataModify
         encJToLBS <$> runQueryM env source query
@@ -667,6 +671,9 @@ execQuery env httpManager metadata queryBs = runExceptT do
 
 instance Tracing.HasReporter ServerAppM
 
+instance HasResolveCustomSource ServerAppM where
+  askResolveCustomSource = pure defaultResolveCustomSource
+
 instance MonadQueryInstrumentation ServerAppM where
   askInstrumentQuery _ = pure (id, noProfile)
 

server/src-lib/Hasura/RQL/DDL/Metadata.hs

@@ -69,7 +69,8 @@ runClearMetadata _ = do
       Nothing -> emptyMetadata
       Just defaultSourceMetadata ->
         let emptyDefaultSource = SourceMetadata defaultSource mempty mempty
-                                 $ _smConfiguration defaultSourceMetadata
+                                   (_smConfiguration defaultSourceMetadata)
+                                   (_smReplicas defaultSourceMetadata)
         in emptyMetadata
            & metaSources %~ HM.insert defaultSource emptyDefaultSource
   pure successMsg
@@ -405,7 +406,7 @@ fetchMetadataFromHdbTables defaultSourceConfig = liftTx do
 
   let tableMetadatas = mapFromL _tmTable $ HM.elems postRelMap
       sources = HM.singleton defaultSource $
-                SourceMetadata defaultSource tableMetadatas functions defaultSourceConfig
+                SourceMetadata defaultSource tableMetadatas functions defaultSourceConfig [] {- TODO: not sure what to do here -}
   pure $ Metadata sources remoteSchemas collections
                   allowlist customTypes actions cronTriggers
 

server/src-lib/Hasura/RQL/DDL/Schema/Cache.hs

@@ -53,7 +53,8 @@ import           Hasura.RQL.DDL.Schema.Cache.Permission
 import           Hasura.RQL.DDL.Schema.Diff
 import           Hasura.RQL.DDL.Schema.Enum               (fetchEnumValuesFromDb)
 import           Hasura.RQL.DDL.Schema.Function
-import           Hasura.RQL.DDL.Schema.Source             (fetchPgScalars, resolveSource)
+import           Hasura.RQL.DDL.Schema.Source             (HasResolveCustomSource,
+                                                           fetchPgScalars, resolveSource)
 import           Hasura.RQL.DDL.Schema.Table
 import           Hasura.RQL.Types                         hiding (tmTable)
 import           Hasura.Server.Version                    (HasVersion)
@@ -62,6 +63,7 @@ import           Hasura.SQL.Types
 buildRebuildableSchemaCache
   :: ( HasVersion, MonadIO m, MonadError QErr m
      , HasHttpManager m, HasSQLGenCtx m, MonadMetadata m
+     , HasResolveCustomSource m
      )
   => Env.Environment
   -> m RebuildableSchemaCache
@@ -96,7 +98,7 @@ instance (Monad m) => CacheRM (CacheRWT m) where
   askSchemaCache = CacheRWT $ gets (lastBuiltSchemaCache . (^. _1))
 
 instance ( MonadIO m , MonadMetadata m, MonadError QErr m
-         , HasHttpManager m, HasSQLGenCtx m
+         , HasHttpManager m, HasSQLGenCtx m, HasResolveCustomSource m
          ) => CacheRWM (CacheRWT m) where
   buildSchemaCacheWithOptions buildReason invalidations metadataModifier = CacheRWT do
     (RebuildableSchemaCache _ invalidationKeys rule, oldInvalidations, resolvedSources) <- get
@@ -128,7 +130,7 @@ buildSchemaCacheRule
   :: ( HasVersion, ArrowChoice arr, Inc.ArrowDistribute arr, Inc.ArrowCache m arr
      , MonadIO m, MonadUnique m, MonadError QErr m
      , MonadReader BuildContext m, HasHttpManager m, HasSQLGenCtx m
-     , MonadBaseControl IO m
+     , MonadBaseControl IO m, HasResolveCustomSource m
      )
   => Env.Environment
   -> (Metadata, InvalidationKeys) `arr` SchemaCache
@@ -193,6 +195,7 @@ buildSchemaCacheRule env = proc (metadata, invalidationKeys) -> do
          , ArrowWriter (Seq CollectedInfo) arr
          , MonadIO m, MonadBaseControl IO m
          , MonadReader BuildContext m
+         , HasResolveCustomSource m
          )
       => ( Inc.Dependency (HashMap SourceName Inc.InvalidationKey)
          , SourceMetadata
@@ -203,14 +206,14 @@ buildSchemaCacheRule env = proc (metadata, invalidationKeys) -> do
       Inc.dependOn -< Inc.selectKeyD sourceName invalidationKeys
       (| withRecordInconsistency (
            liftEitherA <<< bindA -< (getResolvedSourceFromBuildContext sourceName <$> ask) >>= \case
-               Nothing -> resolveSource env $ _smConfiguration sourceMetadata
+               Nothing -> resolveSource env (_smConfiguration sourceMetadata) (_smReplicas sourceMetadata)
                Just rs -> pure $ Right rs)
        |) metadataObj
 
     buildSource
       :: ( ArrowChoice arr, Inc.ArrowDistribute arr, Inc.ArrowCache m arr
          , ArrowWriter (Seq CollectedInfo) arr, MonadReader BuildContext m
-         , HasSQLGenCtx m, MonadIO m, MonadError QErr m)
+         , HasSQLGenCtx m, MonadIO m, MonadError QErr m, HasResolveCustomSource m)
       => ( SourceMetadata
          , PGSourceConfig
          , PostgresTablesMetadata
@@ -220,7 +223,7 @@ buildSchemaCacheRule env = proc (metadata, invalidationKeys) -> do
          , Inc.Dependency InvalidationKeys
          ) `arr` SourceOutput
     buildSource = proc (sourceMetadata, sourceConfig, pgTables, pgFunctions, enumTables, remoteSchemaMap, invalidationKeys) -> do
-      let SourceMetadata source tables functions _ = sourceMetadata
+      let SourceMetadata source tables functions _ _ = sourceMetadata
           (tableInputs, nonColumnInputs, permissions) = unzip3 $ map mkTableInputs $ M.elems tables
           eventTriggers = map (_tmTable &&& (M.elems . _tmEventTriggers)) (M.elems tables)
           -- HashMap k a -> HashMap k b -> HashMap k (a, b)
@@ -281,7 +284,7 @@ buildSchemaCacheRule env = proc (metadata, invalidationKeys) -> do
       :: ( ArrowChoice arr, Inc.ArrowDistribute arr, Inc.ArrowCache m arr
          , ArrowWriter (Seq CollectedInfo) arr, MonadIO m, MonadBaseControl IO m
          , MonadError QErr m, MonadUnique m, MonadReader BuildContext m
-         , HasHttpManager m, HasSQLGenCtx m )
+         , HasHttpManager m, HasSQLGenCtx m, HasResolveCustomSource m )
       => (Metadata, Inc.Dependency InvalidationKeys) `arr` BuildOutputs
     buildAndCollectInfo = proc (metadata, invalidationKeys) -> do
       let Metadata sources remoteSchemas collections allowlists

server/src-lib/Hasura/RQL/DDL/Schema/Cache/Common.hs

@@ -20,6 +20,7 @@ import           Control.Monad.Unique
 
 import qualified Hasura.Incremental           as Inc
 
+import           Hasura.RQL.DDL.Schema.Source
 import           Hasura.RQL.Types
 import           Hasura.SQL.Types
 
@@ -124,8 +125,9 @@ $(makeLenses ''BuildOutputs)
 -- | Parameters required for schema cache build
 data CacheBuildParams
   = CacheBuildParams
-  { _cbpManager   :: !HTTP.Manager
-  , _cbpSqlGenCtx :: !SQLGenCtx
+  { _cbpManager       :: !HTTP.Manager
+  , _cbpSqlGenCtx     :: !SQLGenCtx
+  , _cbpRslvCustomSrc :: !ResolveCustomSource
   }
 
 -- | The monad in which @'RebuildableSchemaCache' is being run
@@ -145,18 +147,23 @@ instance HasHttpManager CacheBuild where
 
 instance HasSQLGenCtx CacheBuild where
   askSQLGenCtx = asks _cbpSqlGenCtx
+
+instance HasResolveCustomSource CacheBuild where
+  askResolveCustomSource = asks _cbpRslvCustomSrc
 
 runCacheBuild
   :: ( MonadIO m
      , MonadError QErr m
      , HasHttpManager m
      , HasSQLGenCtx m
+     , HasResolveCustomSource m
      )
   => CacheBuild a -> m a
 runCacheBuild (CacheBuild m) = do
   httpManager <- askHttpManager
   sqlGenCtx   <- askSQLGenCtx
-  let params = CacheBuildParams httpManager sqlGenCtx
+  rslvCustomSrc <- askResolveCustomSource
+  let params = CacheBuildParams httpManager sqlGenCtx rslvCustomSrc
   liftEitherM $ liftIO $ runExceptT (runReaderT m params)
 
 data RebuildableSchemaCache

server/src-lib/Hasura/RQL/DDL/Schema/Source.hs

@@ -8,26 +8,53 @@ import           Hasura.RQL.DDL.Deps
 import           Hasura.RQL.DDL.Schema.Common
 import           Hasura.RQL.Types
 import           Hasura.SQL.Types
+import           Hasura.Tracing
 
 import qualified Data.Environment             as Env
 import qualified Data.HashMap.Strict          as HM
 import qualified Data.HashSet                 as S
 import qualified Database.PG.Query            as Q
 
-resolveSource
-  :: (MonadIO m, MonadBaseControl IO m)
-  => Env.Environment -> SourceConfiguration -> m (Either QErr ResolvedSource)
-resolveSource env config = runExceptT do
-  let SourceConfiguration urlConf connSettings = config
-      SourceConnSettings maxConns idleTimeout retries = connSettings
+type ResolveCustomSource =
+     Env.Environment 
+  -> SourceConfiguration
+  -> [SourceConfiguration]
+  -> IO (Either QErr PGSourceConfig)
+
+class Monad m => HasResolveCustomSource m where
+  askResolveCustomSource :: m ResolveCustomSource
+
+  -- A default for monad transformer instances
+  default askResolveCustomSource
+    :: (m ~ t n, MonadTrans t, HasResolveCustomSource n)
+    => m ResolveCustomSource
+  askResolveCustomSource = lift askResolveCustomSource
+
+instance HasResolveCustomSource m => HasResolveCustomSource (ReaderT r m)
+instance HasResolveCustomSource m => HasResolveCustomSource (ExceptT e m)
+instance HasResolveCustomSource m => HasResolveCustomSource (TraceT m)
+
+defaultResolveCustomSource :: ResolveCustomSource
+defaultResolveCustomSource env (SourceConfiguration urlConf connSettings) _replicas = runExceptT do
+  let SourceConnSettings maxConns idleTimeout retries = connSettings
   urlText <- resolveUrlConf env urlConf
   let connInfo = Q.ConnInfo retries $ Q.CDDatabaseURI $ txtToBs urlText
       connParams = Q.defaultConnParams{ Q.cpIdleTime = idleTimeout
                                       , Q.cpConns = maxConns
                                       }
   pgPool <- liftIO $ Q.initPGPool connInfo connParams (\_ -> pure ()) -- FIXME? Pg logger
   let pgExecCtx = mkPGExecCtx Q.ReadCommitted pgPool
-      sourceConfig  = PGSourceConfig pgExecCtx connInfo
+  pure $ PGSourceConfig pgExecCtx connInfo
+
+resolveSource
+  :: (MonadIO m, MonadBaseControl IO m, HasResolveCustomSource m)
+  => Env.Environment 
+  -> SourceConfiguration 
+  -> [SourceConfiguration]
+  -> m (Either QErr ResolvedSource)
+resolveSource env config replicas = runExceptT do
+  resolver <- askResolveCustomSource
+  sourceConfig <- ExceptT . liftIO $ resolver env config replicas
 
   (tablesMeta, functionsMeta, pgScalars) <- runLazyTx (_pscExecCtx sourceConfig) Q.ReadWrite $ do
     initSource

server/src-lib/Hasura/RQL/Types/Metadata.hs

@@ -311,6 +311,7 @@ data SourceMetadata
   , _smTables        :: !Tables
   , _smFunctions     :: !Functions
   , _smConfiguration :: !SourceConfiguration
+  , _smReplicas      :: ![SourceConfiguration]
   } deriving (Show, Eq, Lift, Generic)
 instance Cacheable SourceMetadata
 $(makeLenses ''SourceMetadata)
@@ -320,12 +321,15 @@ instance FromJSON SourceMetadata where
     _smTables        <- mapFromL _tmTable <$> o .: "tables"
     _smFunctions     <- mapFromL _fmFunction <$> o .:? "functions" .!= []
     _smConfiguration <- o .: "configuration"
+    _smReplicas      <- o .:? "read_replicas" .!= []
     pure SourceMetadata{..}
 
 mkSourceMetadata
   :: SourceName -> UrlConf -> SourceConnSettings -> SourceMetadata
 mkSourceMetadata name urlConf connSettings =
-  SourceMetadata name mempty mempty $ SourceConfiguration urlConf connSettings
+  SourceMetadata name mempty mempty 
+    (SourceConfiguration urlConf connSettings)
+    mempty
 
 type Sources = M.HashMap SourceName SourceMetadata
 
@@ -464,7 +468,8 @@ metadataToOrdJSON ( Metadata
           tablesPair = ("tables", AO.array $ map tableMetaToOrdJSON $ M.elems _smTables)
           functionsPair = listToMaybeOrdPair "functions" functionMetadataToOrdJSON $ M.elems _smFunctions
           configurationPair = [("configuration", AO.toOrdered _smConfiguration)]
-      in AO.object $ [sourceNamePair, tablesPair] <> maybeToList functionsPair <> configurationPair
+          replicasPair = [("read_replicas", AO.toOrdered _smReplicas) | not (null _smReplicas)]
+      in AO.object $ [sourceNamePair, tablesPair] <> maybeToList functionsPair <> configurationPair <> replicasPair
 
     tableMetaToOrdJSON :: TableMetadata -> AO.Value
     tableMetaToOrdJSON ( TableMetadata

server/src-lib/Hasura/RQL/Types/Run.hs

@@ -22,6 +22,7 @@ import           Control.Monad.Trans.Control (MonadBaseControl)
 import           Control.Monad.Unique
 import           Data.Aeson
 
+import           Hasura.RQL.DDL.Schema.Source
 import           Hasura.RQL.Types
 import qualified Hasura.Tracing              as Tracing
 
@@ -30,6 +31,7 @@ data RunCtx
   { _rcUserInfo  :: !UserInfo
   , _rcHttpMgr   :: !HTTP.Manager
   , _rcSqlGenCtx :: !SQLGenCtx
+  , _rcRslvCustomSrc   :: !ResolveCustomSource
   }
 
 newtype BaseRunT m a
@@ -58,6 +60,9 @@ instance (Monad m) => MonadMetadata (BaseRunT m) where
   fetchMetadata = get
   updateMetadata = put
 
+instance (Monad m) => HasResolveCustomSource (BaseRunT m) where
+  askResolveCustomSource = asks _rcRslvCustomSrc
+
 runBaseRunT :: RunCtx -> Metadata -> BaseRunT m a -> m (a, Metadata)
 runBaseRunT runCtx metadata (BaseRunT m) =
   runStateT (runReaderT m runCtx) metadata
@@ -73,6 +78,7 @@ newtype MetadataRun m a
            , UserInfoM
            , HasHttpManager
            , HasSQLGenCtx
+           , HasResolveCustomSource
            )
 
 runInMetadataRun :: (Monad m) => MetadataStorageT m a -> MetadataRun m a
@@ -117,6 +123,7 @@ newtype QueryRun a
            , UserInfoM
            , HasHttpManager
            , HasSQLGenCtx
+           , HasResolveCustomSource
            )
 
 peelQueryRun

server/src-lib/Hasura/Server/API/Metadata.hs

@@ -146,15 +146,16 @@ runMetadataRequest
   -> UserInfo
   -> HTTP.Manager
   -> SQLGenCtx
+  -> ResolveCustomSource
   -> RebuildableSchemaCache
   -> Metadata
   -> RQLMetadata
   -> m (EncJSON, MetadataStateResult)
-runMetadataRequest env userInfo httpManager sqlGenCtx schemaCache metadata request = do
+runMetadataRequest env userInfo httpManager sqlGenCtx srcResolver schemaCache metadata request = do
   ((r, modSchemaCache, cacheInvalidations), modMetadata) <-
     runMetadataRequestM env request
     & runCacheRWT schemaCache
-    & peelMetadataRun (RunCtx userInfo httpManager sqlGenCtx) metadata
+    & peelMetadataRun (RunCtx userInfo httpManager sqlGenCtx srcResolver) metadata
     & runExceptT
     & liftEitherM
   pure (r, MetadataStateResult modSchemaCache cacheInvalidations modMetadata)