server: implement websocket compression setting (fixes #3292)

CHANGELOG.md

@@ -67,6 +67,7 @@ This release contains the [PDV refactor (#4111)](https://github.com/hasura/graph
 
 (Add entries here in the order of: server, console, cli, docs, others)
 
+- server: add `--websocket-compression` command-line flag for enabling websocket compression (fix #3292)
 - server: some mutations that cannot be performed will no longer be in the schema (for instance, `delete_by_pk` mutations won't be shown to users that do not have select permissions on all primary keys) (#4111)
 - server: miscellaneous description changes (#4111)
 - server: treat the absence of `backend_only` configuration and `backend_only: false` equally (closing #5059) (#4111)

server/src-lib/Hasura/App.hs

@@ -363,6 +363,7 @@ runHGEServer env ServeOptions{..} InitCtx{..} pgExecCtx initTime shutdownApp pos
              postPollHook
              _icSchemaCache
              ekgStore
+             soConnectionOptions
 
   -- log inconsistent schema objects
   inconsObjs <- scInconsistentObjs <$> liftIO (getSCFromRef cacheRef)

server/src-lib/Hasura/Server/App.hs

@@ -603,9 +603,10 @@ mkWaiApp
   -> Maybe EL.LiveQueryPostPollHook
   -> (RebuildableSchemaCache Run, Maybe UTCTime)
   -> EKG.Store
+  -> WS.ConnectionOptions
   -> m HasuraApp
 mkWaiApp env isoLevel logger sqlGenCtx enableAL pool pgExecCtxCustom ci httpManager mode corsCfg enableConsole consoleAssetsDir
-         enableTelemetry instanceId apis lqOpts _ {- planCacheOptions -} responseErrorsConfig liveQueryHook (schemaCache, cacheBuiltTime) ekgStore  = do
+         enableTelemetry instanceId apis lqOpts _ {- planCacheOptions -} responseErrorsConfig liveQueryHook (schemaCache, cacheBuiltTime) ekgStore connectionOptions  = do
 
     -- See Note [Temporarily disabling query plan caching]
     -- (planCache, schemaCacheRef) <- initialiseCache
@@ -646,7 +647,7 @@ mkWaiApp env isoLevel logger sqlGenCtx enableAL pool pgExecCtxCustom ci httpMana
         stopWSServer = WS.stopWSServerApp wsServerEnv
 
     waiApp <- liftWithStateless $ \lowerIO ->
-      pure $ WSC.websocketsOr WS.defaultConnectionOptions (\ip conn -> lowerIO $ wsServerApp ip conn) spockApp
+      pure $ WSC.websocketsOr connectionOptions (\ip conn -> lowerIO $ wsServerApp ip conn) spockApp
 
     return $ HasuraApp waiApp schemaCacheRef cacheBuiltTime stopWSServer
   where

server/src-lib/Hasura/Server/Init.hs

@@ -20,6 +20,7 @@ import qualified Text.PrettyPrint.ANSI.Leijen     as PP
 import           Data.FileEmbed                   (embedStringFile)
 import           Data.Time                        (NominalDiffTime)
 import           Network.Wai.Handler.Warp         (HostPreference)
+import qualified Network.WebSockets               as WS
 import           Options.Applicative
 
 import qualified Hasura.Cache.Bounded             as Cache
@@ -170,12 +171,22 @@ mkServeOptions rso = do
   eventsFetchInterval <- withEnv (rsoEventsFetchInterval rso) (fst eventsFetchIntervalEnv)
   logHeadersFromEnv <- withEnvBool (rsoLogHeadersFromEnv rso) (fst logHeadersFromEnvEnv)
 
+  webSocketCompressionFromEnv <- withEnvBool (rsoWebSocketCompression rso) $
+                                 fst webSocketCompressionEnv
+
+  let connectionOptions = WS.defaultConnectionOptions {
+                            WS.connectionCompressionOptions =
+                              if webSocketCompressionFromEnv
+                                then WS.PermessageDeflateCompression WS.defaultPermessageDeflate
+                                else WS.NoCompression
+                          }
+
   return $ ServeOptions port host connParams txIso adminScrt authHook jwtSecret
                         unAuthRole corsCfg enableConsole consoleAssetsDir
                         enableTelemetry strfyNum enabledAPIs lqOpts enableAL
                         enabledLogs serverLogLevel planCacheOptions
                         internalErrorsConfig eventsHttpPoolSize eventsFetchInterval
-                        logHeadersFromEnv
+                        logHeadersFromEnv connectionOptions
   where
 #ifdef DeveloperAPIs
     defaultAPIs = [METADATA,GRAPHQL,PGDUMP,CONFIG,DEVELOPER]
@@ -931,6 +942,7 @@ serveOptsToLog so =
       , "enabled_log_types" J..= soEnabledLogTypes so
       , "log_level" J..= soLogLevel so
       , "plan_cache_options" J..= soPlanCacheOptions so
+      , "websocket_compression_options" J..= show (WS.connectionCompressionOptions . soConnectionOptions $ so)
       ]
 
 mkGenericStrLog :: L.LogLevel -> T.Text -> String -> StartupLog
@@ -976,6 +988,7 @@ serveOptionsParser =
   <*> parseGraphqlEventsHttpPoolSize
   <*> parseGraphqlEventsFetchInterval
   <*> parseLogHeadersFromEnv
+  <*> parseWebSocketCompression
 
 -- | This implements the mapping between application versions
 -- and catalog schema versions.
@@ -1010,3 +1023,15 @@ downgradeOptionsParser =
         ( long ("to-" <> v) <>
           help ("Downgrade to graphql-engine version " <> v <> " (equivalent to --to-catalog-version " <> catalogVersion <> ")")
         )
+
+webSocketCompressionEnv :: (String, String)
+webSocketCompressionEnv =
+  ( "HASURA_GRAPHQL_CONNECTION_COMPRESSION"
+  , "Enable WebSocket permessage-deflate compression (default: false)"
+  )
+
+parseWebSocketCompression :: Parser Bool
+parseWebSocketCompression =
+  switch ( long "websocket-compression" <>
+           help (snd webSocketCompressionEnv)
+         )

server/src-lib/Hasura/Server/Init/Config.hs

@@ -12,6 +12,7 @@ import qualified Database.PG.Query                as Q
 import           Data.Char                        (toLower)
 import           Data.Time
 import           Network.Wai.Handler.Warp         (HostPreference)
+import qualified Network.WebSockets               as WS
 
 import qualified Hasura.Cache.Bounded             as Cache
 import qualified Hasura.GraphQL.Execute.LiveQuery as LQ
@@ -38,32 +39,33 @@ type RawAuthHook = AuthHookG (Maybe T.Text) (Maybe AuthHookType)
 
 data RawServeOptions impl
   = RawServeOptions
-  { rsoPort                :: !(Maybe Int)
-  , rsoHost                :: !(Maybe HostPreference)
-  , rsoConnParams          :: !RawConnParams
-  , rsoTxIso               :: !(Maybe Q.TxIsolation)
-  , rsoAdminSecret         :: !(Maybe AdminSecretHash)
-  , rsoAuthHook            :: !RawAuthHook
-  , rsoJwtSecret           :: !(Maybe JWTConfig)
-  , rsoUnAuthRole          :: !(Maybe RoleName)
-  , rsoCorsConfig          :: !(Maybe CorsConfig)
-  , rsoEnableConsole       :: !Bool
-  , rsoConsoleAssetsDir    :: !(Maybe Text)
-  , rsoEnableTelemetry     :: !(Maybe Bool)
-  , rsoWsReadCookie        :: !Bool
-  , rsoStringifyNum        :: !Bool
-  , rsoEnabledAPIs         :: !(Maybe [API])
-  , rsoMxRefetchInt        :: !(Maybe LQ.RefetchInterval)
-  , rsoMxBatchSize         :: !(Maybe LQ.BatchSize)
-  , rsoEnableAllowlist     :: !Bool
-  , rsoEnabledLogTypes     :: !(Maybe [L.EngineLogType impl])
-  , rsoLogLevel            :: !(Maybe L.LogLevel)
-  , rsoPlanCacheSize       :: !(Maybe Cache.CacheSize)
-  , rsoDevMode             :: !Bool
-  , rsoAdminInternalErrors :: !(Maybe Bool)
-  , rsoEventsHttpPoolSize  :: !(Maybe Int)
-  , rsoEventsFetchInterval :: !(Maybe Milliseconds)
-  , rsoLogHeadersFromEnv   :: !Bool
+  { rsoPort                  :: !(Maybe Int)
+  , rsoHost                  :: !(Maybe HostPreference)
+  , rsoConnParams            :: !RawConnParams
+  , rsoTxIso                 :: !(Maybe Q.TxIsolation)
+  , rsoAdminSecret           :: !(Maybe AdminSecretHash)
+  , rsoAuthHook              :: !RawAuthHook
+  , rsoJwtSecret             :: !(Maybe JWTConfig)
+  , rsoUnAuthRole            :: !(Maybe RoleName)
+  , rsoCorsConfig            :: !(Maybe CorsConfig)
+  , rsoEnableConsole         :: !Bool
+  , rsoConsoleAssetsDir      :: !(Maybe Text)
+  , rsoEnableTelemetry       :: !(Maybe Bool)
+  , rsoWsReadCookie          :: !Bool
+  , rsoStringifyNum          :: !Bool
+  , rsoEnabledAPIs           :: !(Maybe [API])
+  , rsoMxRefetchInt          :: !(Maybe LQ.RefetchInterval)
+  , rsoMxBatchSize           :: !(Maybe LQ.BatchSize)
+  , rsoEnableAllowlist       :: !Bool
+  , rsoEnabledLogTypes       :: !(Maybe [L.EngineLogType impl])
+  , rsoLogLevel              :: !(Maybe L.LogLevel)
+  , rsoPlanCacheSize         :: !(Maybe Cache.CacheSize)
+  , rsoDevMode               :: !Bool
+  , rsoAdminInternalErrors   :: !(Maybe Bool)
+  , rsoEventsHttpPoolSize    :: !(Maybe Int)
+  , rsoEventsFetchInterval   :: !(Maybe Milliseconds)
+  , rsoLogHeadersFromEnv     :: !Bool
+  , rsoWebSocketCompression :: !Bool
   }
 
 -- | @'ResponseInternalErrorsConfig' represents the encoding of the internal
@@ -106,6 +108,7 @@ data ServeOptions impl
   , soEventsHttpPoolSize           :: !(Maybe Int)
   , soEventsFetchInterval          :: !(Maybe Milliseconds)
   , soLogHeadersFromEnv            :: !Bool
+  , soConnectionOptions            :: !WS.ConnectionOptions
   }
 
 data DowngradeOptions