Websocket test

composition/src/directive-definition-data/directive-definition-data.ts

@@ -25,6 +25,7 @@ import {
   EDFS_NATS_STREAM_CONFIGURATION,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   ENUM_UPPER,
   ENUM_VALUE_UPPER,
@@ -104,6 +105,7 @@ import {
   EDFS_NATS_REQUEST_DEFINITION,
   EDFS_NATS_SUBSCRIBE_DEFINITION,
   EDFS_REDIS_PUBLISH_DEFINITION,
+  EDFS_PUSHER_SUBSCRIBE_DEFINITION,
   EDFS_REDIS_SUBSCRIBE_DEFINITION,
   EXTENDS_DEFINITION,
   EXTERNAL_DEFINITION,
@@ -804,6 +806,44 @@ export const REDIS_SUBSCRIBE_DEFINITION_DATA = newDirectiveDefinitionData({
   requiredArgumentNames: new Set<ArgumentName>([CHANNELS]),
 });
 
+export const PUSHER_SUBSCRIBE_DEFINITION_DATA = newDirectiveDefinitionData({
+  argumentDataByName: new Map<ArgumentName, DirectiveArgumentData>([
+    [
+      CHANNELS,
+      newDirectiveArgumentData({
+        directive: `@${EDFS_PUSHER_SUBSCRIBE}`,
+        name: CHANNELS,
+        namedTypeKind: Kind.SCALAR_TYPE_DEFINITION,
+        typeNode: {
+          kind: Kind.NON_NULL_TYPE,
+          type: {
+            kind: Kind.LIST_TYPE,
+            type: REQUIRED_STRING_TYPE_NODE,
+          },
+        },
+      }),
+    ],
+    [
+      PROVIDER_ID,
+      newDirectiveArgumentData({
+        directive: `@${EDFS_PUSHER_SUBSCRIBE}`,
+        name: PROVIDER_ID,
+        namedTypeKind: Kind.SCALAR_TYPE_DEFINITION,
+        typeNode: REQUIRED_STRING_TYPE_NODE,
+        defaultValue: {
+          kind: Kind.STRING,
+          value: DEFAULT_EDFS_PROVIDER_ID,
+        },
+      }),
+    ],
+  ]),
+  locations: new Set<DirectiveLocation>([FIELD_DEFINITION_UPPER]),
+  name: EDFS_PUSHER_SUBSCRIBE,
+  node: EDFS_PUSHER_SUBSCRIBE_DEFINITION,
+  optionalArgumentNames: new Set<ArgumentName>([PROVIDER_ID]),
+  requiredArgumentNames: new Set<ArgumentName>([CHANNELS]),
+});
+
 export const REQUIRE_FETCH_REASONS_DEFINITION_DATA = newDirectiveDefinitionData({
   isRepeatable: true,
   locations: new Set<DirectiveLocation>([FIELD_DEFINITION_UPPER, INTERFACE_UPPER, OBJECT_UPPER]),

composition/src/router-configuration/types.ts

@@ -13,6 +13,8 @@ export type KafkaEventType = 'subscribe' | 'publish';
 
 export type RedisEventType = 'subscribe' | 'publish';
 
+export type PusherEventType = 'subscribe';
+
 export type StreamConfiguration = {
   consumerInactiveThreshold: number;
   consumerName: string;
@@ -44,7 +46,19 @@ export type RedisEventConfiguration = {
   type: RedisEventType;
 };
 
-export type EventConfiguration = KafkaEventConfiguration | NatsEventConfiguration | RedisEventConfiguration;
+export type PusherEventConfiguration = {
+  fieldName: string;
+  providerId: string;
+  providerType: 'pusher';
+  channels: string[];
+  type: PusherEventType;
+};
+
+export type EventConfiguration =
+  | KafkaEventConfiguration
+  | NatsEventConfiguration
+  | RedisEventConfiguration
+  | PusherEventConfiguration;
 
 export type SubscriptionFilterValue = boolean | null | number | string;
 

composition/src/utils/string-constants.ts

@@ -41,6 +41,7 @@ export const EDFS_PUBLISH_RESULT = 'edfs__PublishResult';
 export const EDFS_NATS_STREAM_CONFIGURATION = 'edfs__NatsStreamConfiguration';
 export const EDFS_REDIS_PUBLISH = 'edfs__redisPublish';
 export const EDFS_REDIS_SUBSCRIBE = 'edfs__redisSubscribe';
+export const EDFS_PUSHER_SUBSCRIBE = 'edfs__pusherSubscribe';
 export const ENTITIES = 'entities';
 export const ENTITIES_FIELD = '_entities';
 export const OPENFED_ENTITY_CACHE = 'openfed__entityCache';
@@ -103,6 +104,7 @@ export const PROPAGATE = 'propagate';
 export const PROVIDER_TYPE_KAFKA = 'kafka';
 export const PROVIDER_TYPE_NATS = 'nats';
 export const PROVIDER_TYPE_REDIS = 'redis';
+export const PROVIDER_TYPE_PUSHER = 'pusher';
 export const NOT_APPLICABLE = 'N/A';
 export const NAME = 'name';
 export const NEGATIVE_CACHE_TTL = 'negativeCacheTTL';

composition/src/v1/constants/constants.ts

@@ -17,6 +17,7 @@ import {
   EDFS_NATS_REQUEST,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   EXTENDS,
   EXTERNAL,
@@ -58,6 +59,7 @@ import {
   EDFS_NATS_REQUEST_DEFINITION,
   EDFS_NATS_SUBSCRIBE_DEFINITION,
   EDFS_REDIS_PUBLISH_DEFINITION,
+  EDFS_PUSHER_SUBSCRIBE_DEFINITION,
   EDFS_REDIS_SUBSCRIBE_DEFINITION,
   EXTENDS_DEFINITION,
   EXTERNAL_DEFINITION,
@@ -103,6 +105,7 @@ export const DIRECTIVE_DEFINITION_BY_NAME: ReadonlyMap<DirectiveName, DirectiveD
   [EDFS_NATS_SUBSCRIBE, EDFS_NATS_SUBSCRIBE_DEFINITION],
   [EDFS_REDIS_PUBLISH, EDFS_REDIS_PUBLISH_DEFINITION],
   [EDFS_REDIS_SUBSCRIBE, EDFS_REDIS_SUBSCRIBE_DEFINITION],
+  [EDFS_PUSHER_SUBSCRIBE, EDFS_PUSHER_SUBSCRIBE_DEFINITION],
   [EXTENDS, EXTENDS_DEFINITION],
   [EXTERNAL, EXTERNAL_DEFINITION],
   [INACCESSIBLE, INACCESSIBLE_DEFINITION],

composition/src/v1/constants/directive-definitions.ts

@@ -25,6 +25,7 @@ import {
   EDFS_NATS_STREAM_CONFIGURATION,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   ENUM_UPPER,
   ENUM_VALUE_UPPER,
@@ -488,6 +489,36 @@ export const EDFS_REDIS_SUBSCRIBE_DEFINITION: DirectiveDefinitionNode = {
   repeatable: false,
 };
 
+// directive @edfs__pusherSubscribe(channels: [String!]!, providerId: String! = "default") on FIELD_DEFINITION
+export const EDFS_PUSHER_SUBSCRIBE_DEFINITION: DirectiveDefinitionNode = {
+  arguments: [
+    {
+      kind: Kind.INPUT_VALUE_DEFINITION,
+      name: stringToNameNode(CHANNELS),
+      type: {
+        kind: Kind.NON_NULL_TYPE,
+        type: {
+          kind: Kind.LIST_TYPE,
+          type: REQUIRED_STRING_TYPE_NODE,
+        },
+      },
+    },
+    {
+      kind: Kind.INPUT_VALUE_DEFINITION,
+      name: stringToNameNode(PROVIDER_ID),
+      type: REQUIRED_STRING_TYPE_NODE,
+      defaultValue: {
+        kind: Kind.STRING,
+        value: DEFAULT_EDFS_PROVIDER_ID,
+      },
+    },
+  ],
+  kind: Kind.DIRECTIVE_DEFINITION,
+  locations: [stringToNameNode(FIELD_DEFINITION_UPPER)],
+  name: stringToNameNode(EDFS_PUSHER_SUBSCRIBE),
+  repeatable: false,
+};
+
 export const INACCESSIBLE_DEFINITION: DirectiveDefinitionNode = {
   kind: Kind.DIRECTIVE_DEFINITION,
   locations: stringArrayToNameNodeArray([

composition/src/v1/constants/strings.ts

@@ -14,6 +14,7 @@ import {
   EDFS_NATS_STREAM_CONFIGURATION,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   ENUM_UPPER,
   ENUM_VALUE_UPPER,
@@ -84,6 +85,7 @@ export const EVENT_DIRECTIVE_NAMES: ReadonlySet<DirectiveName> = new Set<Directi
   EDFS_NATS_REQUEST,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
 ]);
 

composition/src/v1/normalization/normalization-factory.ts

@@ -294,6 +294,7 @@ import {
   EDFS_NATS_SUBSCRIBE,
   EDFS_PUBLISH_RESULT,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   ENTITIES_FIELD,
   EXTENDS,
@@ -331,6 +332,7 @@ import {
   PROVIDER_ID,
   PROVIDER_TYPE_KAFKA,
   PROVIDER_TYPE_NATS,
+  PROVIDER_TYPE_PUSHER,
   PROVIDER_TYPE_REDIS,
   PUBLISH,
   QUERY,
@@ -3410,6 +3412,54 @@ export class NormalizationFactory {
     };
   }
 
+  getPusherSubscribeConfiguration(
+    directive: ConstDirectiveNode,
+    argumentDataByArgumentName: Map<string, InputValueData>,
+    fieldName: string,
+    errorMessages: string[],
+  ): EventConfiguration | undefined {
+    const channels: string[] = [];
+    let providerId = DEFAULT_EDFS_PROVIDER_ID;
+    for (const argumentNode of directive.arguments || []) {
+      switch (argumentNode.name.value) {
+        case CHANNELS: {
+          //@TODO list coercion
+          if (argumentNode.value.kind !== Kind.LIST) {
+            errorMessages.push(invalidEventSubjectsErrorMessage(CHANNELS));
+            continue;
+          }
+          for (const value of argumentNode.value.values) {
+            if (value.kind !== Kind.STRING || value.value.length < 1) {
+              errorMessages.push(invalidEventSubjectsItemErrorMessage(CHANNELS));
+              break;
+            }
+            validateArgumentTemplateReferences(value.value, argumentDataByArgumentName, errorMessages);
+            channels.push(value.value);
+          }
+          break;
+        }
+        case PROVIDER_ID: {
+          if (argumentNode.value.kind !== Kind.STRING || argumentNode.value.value.length < 1) {
+            errorMessages.push(invalidEventProviderIdErrorMessage);
+            continue;
+          }
+          providerId = argumentNode.value.value;
+          break;
+        }
+      }
+    }
+    if (errorMessages.length > 0) {
+      return;
+    }
+    return {
+      fieldName,
+      providerId,
+      providerType: PROVIDER_TYPE_PUSHER,
+      channels,
+      type: SUBSCRIBE,
+    };
+  }
+
   validateSubscriptionFilterDirectiveLocation(node: FieldDefinitionNode) {
     if (!node.directives) {
       return;
@@ -3505,6 +3555,15 @@ export class NormalizationFactory {
           );
           break;
         }
+        case EDFS_PUSHER_SUBSCRIBE: {
+          eventConfiguration = this.getPusherSubscribeConfiguration(
+            directive,
+            argumentDataByArgumentName,
+            fieldName,
+            errorMessages,
+          );
+          break;
+        }
         default:
           continue;
       }
@@ -3534,7 +3593,7 @@ export class NormalizationFactory {
       case OperationTypeNode.QUERY:
         return new Set<string>([EDFS_NATS_REQUEST]);
       case OperationTypeNode.SUBSCRIPTION:
-        return new Set<string>([EDFS_KAFKA_SUBSCRIBE, EDFS_NATS_SUBSCRIBE, EDFS_REDIS_SUBSCRIBE]);
+        return new Set<string>([EDFS_KAFKA_SUBSCRIBE, EDFS_NATS_SUBSCRIBE, EDFS_REDIS_SUBSCRIBE, EDFS_PUSHER_SUBSCRIBE]);
     }
   }
 

composition/src/v1/normalization/utils.ts

@@ -69,6 +69,7 @@ import {
   OVERRIDE_DEFINITION_DATA,
   PROVIDES_DEFINITION_DATA,
   REDIS_PUBLISH_DEFINITION_DATA,
+  PUSHER_SUBSCRIBE_DEFINITION_DATA,
   REDIS_SUBSCRIBE_DEFINITION_DATA,
   REQUIRE_FETCH_REASONS_DEFINITION_DATA,
   REQUIRES_DEFINITION_DATA,
@@ -96,6 +97,7 @@ import {
   EDFS_NATS_REQUEST,
   EDFS_NATS_SUBSCRIBE,
   EDFS_REDIS_PUBLISH,
+  EDFS_PUSHER_SUBSCRIBE,
   EDFS_REDIS_SUBSCRIBE,
   EXTENDS,
   EXTERNAL,
@@ -490,6 +492,7 @@ export function initializeDirectiveDefinitionDatas(): Map<string, DirectiveDefin
     [EDFS_NATS_SUBSCRIBE, NATS_SUBSCRIBE_DEFINITION_DATA],
     [EDFS_REDIS_PUBLISH, REDIS_PUBLISH_DEFINITION_DATA],
     [EDFS_REDIS_SUBSCRIBE, REDIS_SUBSCRIBE_DEFINITION_DATA],
+    [EDFS_PUSHER_SUBSCRIBE, PUSHER_SUBSCRIBE_DEFINITION_DATA],
     [EXTENDS, EXTENDS_DEFINITION_DATA],
     [EXTERNAL, EXTERNAL_DEFINITION_DATA],
     [INACCESSIBLE, INACCESSIBLE_DEFINITION_DATA],

proto/wg/cosmo/node/v1/node.proto

@@ -466,6 +466,11 @@ message RedisEventConfiguration {
   repeated string channels = 2;
 }
 
+message PusherEventConfiguration {
+  EngineEventConfiguration engine_event_configuration = 1;
+  repeated string channels = 2;
+}
+
 message EngineEventConfiguration {
   string provider_id = 1;
   EventType type = 2;
@@ -477,6 +482,7 @@ message DataSourceCustomEvents {
   repeated NatsEventConfiguration nats = 1;
   repeated KafkaEventConfiguration kafka = 2;
   repeated RedisEventConfiguration redis = 3;
+  repeated PusherEventConfiguration pusher = 4;
 }
 
 message DataSourceCustom_Static {

router-tests/go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/redis/go-redis/v9 v9.7.3
 	github.com/sebdah/goldie/v2 v2.7.1
 	github.com/stretchr/testify v1.11.1
+	github.com/tidwall/gjson v1.18.0
 	github.com/twmb/franz-go v1.16.1
 	github.com/twmb/franz-go/pkg/kadm v1.11.0
 	github.com/wundergraph/astjson v1.1.0
@@ -154,7 +155,6 @@ require (
 	github.com/sosodev/duration v1.3.1 // indirect
 	github.com/spf13/cast v1.7.1 // indirect
 	github.com/stretchr/objx v0.5.3 // indirect
-	github.com/tidwall/gjson v1.18.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/tidwall/sjson v1.2.5 // indirect

router-tests/modules/custom-set-wildcard-scope/module.go

@@ -0,0 +1,28 @@
+package custom_set_wildcard_scope
+
+import (
+	"net/http"
+
+	"github.com/wundergraph/cosmo/router/core"
+)
+
+const myModuleID = "setWildcardScopeModule"
+
+type SetWildcardScopeModule struct{}
+
+func (m *SetWildcardScopeModule) Middleware(ctx core.RequestContext, next http.Handler) {
+	ctx.SetWildcardScope(true)
+	next.ServeHTTP(ctx.ResponseWriter(), ctx.Request())
+}
+
+func (m *SetWildcardScopeModule) Module() core.ModuleInfo {
+	return core.ModuleInfo{
+		ID:       myModuleID,
+		Priority: 2,
+		New: func() core.Module {
+			return &SetWildcardScopeModule{}
+		},
+	}
+}
+
+var _ core.RouterMiddlewareHandler = (*SetWildcardScopeModule)(nil)