feat: add NATS to the router (#333)

Co-authored-by: Aenimus <daviditstutt@gmail.com>
Co-authored-by: Aenimus <47415099+Aenimus@users.noreply.github.com>

Makefile

@@ -122,3 +122,6 @@ docker-build-minikube: docker-build-local
 	minikube image load ghcr.io/wundergraph/cosmo/keycloak:latest & \
 	minikube image load ghcr.io/wundergraph/cosmo/cdn:latest
 	minikube cache reload
+
+run-subgraphs-local:
+	cd demo && go run cmd/all/main.go

composition/src/normalization/normalization-factory.ts

@@ -142,6 +142,9 @@ import {
   ANY_SCALAR,
   ENTITIES_FIELD,
   ENTITY_UNION,
+  EVENTS_PUBLISH,
+  EVENTS_REQUEST,
+  EVENTS_SUBSCRIBE,
   EXTENDS,
   EXTENSIONS,
   EXTERNAL,
@@ -158,9 +161,17 @@ import {
   SCHEMA,
   SERVICE_FIELD,
   SERVICE_OBJECT,
+  SOURCE_ID,
+  TOPIC,
 } from '../utils/string-constants';
 import { buildASTSchema } from '../buildASTSchema/buildASTSchema';
-import { ConfigurationData, ConfigurationDataMap } from '../subgraph/field-configuration';
+import {
+  ConfigurationData,
+  ConfigurationDataMap,
+  EventConfiguration,
+  EventType,
+  RequiredFieldConfiguration,
+} from '../subgraph/field-configuration';
 import { printTypeNode } from '@graphql-tools/merge';
 import { inputValueDefinitionNodeToMutable, MutableInputValueDefinitionNode, ObjectLikeTypeNode } from '../ast/ast';
 import { InternalSubgraph, recordSubgraphName, Subgraph } from '../subgraph/subgraph';
@@ -215,6 +226,7 @@ export class NormalizationFactory {
   entityInterfaces = new Map<string, EntityInterfaceData>();
   extensions: ExtensionMap = new Map<string, ExtensionContainer>();
   isCurrentParentExtension = false;
+  isCurrentParentRootType = false;
   isSubgraphVersionTwo = false;
   fieldSetsByParent = new Map<string, FieldSetContainer>();
   handledRepeatedDirectivesByHostPath = new Map<string, Set<string>>();
@@ -225,6 +237,7 @@ export class NormalizationFactory {
   parents: ParentMap = new Map<string, ParentContainer>();
   parentTypeName = '';
   parentsWithChildArguments = new Set<string>();
+  eventsConfigurations = new Map<string, EventConfiguration[]>();
   overridesByTargetSubgraphName = new Map<string, Map<string, Set<string>>>();
   schemaDefinition: SchemaContainer;
   subgraphName?: string;
@@ -619,6 +632,21 @@ export class NormalizationFactory {
     }
   }
 
+  canContainEventDirectives(): boolean {
+    if (!this.isCurrentParentRootType) {
+      return false;
+    }
+    const operationTypeNode = this.operationTypeNames.get(this.parentTypeName);
+    if (!operationTypeNode) {
+      return ROOT_TYPES.has(this.parentTypeName);
+    }
+    return (
+      operationTypeNode === OperationTypeNode.QUERY ||
+      operationTypeNode === OperationTypeNode.MUTATION ||
+      operationTypeNode === OperationTypeNode.SUBSCRIPTION
+    );
+  }
+
   extractKeyFieldSets(node: ObjectLikeTypeNode, rawFieldSets: Set<string>) {
     const parentTypeName = node.name.value;
     if (!node.directives?.length) {
@@ -824,14 +852,82 @@ export class NormalizationFactory {
     overriddenFieldNamesForParent.add(this.childName);
   }
 
+  extractEventDirectives(node: FieldDefinitionNode) {
+    if (!node.directives) {
+      return;
+    }
+    for (const directive of node.directives) {
+      let eventType: EventType;
+      switch (directive.name.value) {
+        case EVENTS_PUBLISH: {
+          eventType = 'publish';
+          break;
+        }
+        case EVENTS_REQUEST: {
+          eventType = 'request';
+          break;
+        }
+        case EVENTS_SUBSCRIBE: {
+          eventType = 'subscribe';
+          break;
+        }
+        default:
+          continue;
+      }
+      let topic: string | undefined;
+      let sourceId: string | undefined;
+      for (const arg of directive.arguments || []) {
+        if (arg.value.kind !== Kind.STRING) {
+          throw new Error(`Event directive arguments must be strings, ${arg.value.kind} found in argument ${arg.name}`);
+        }
+        switch (arg.name.value) {
+          case TOPIC: {
+            if (topic !== undefined) {
+              throw new Error(`Event directives must have exactly one topic argument, found multiple`);
+            }
+            if (!arg.value.value) {
+              throw new Error(`Event directives must have a non-empty topic argument`);
+            }
+            topic = arg.value.value;
+            break;
+          }
+          case SOURCE_ID: {
+            if (sourceId !== undefined) {
+              throw new Error(`Event directives must have exactly one sourceID argument, found multiple`);
+            }
+            if (!arg.value.value) {
+              throw new Error(`Event directives must have a non-empty sourceID argument`);
+            }
+            sourceId = arg.value.value;
+            break;
+          }
+          default:
+            throw new Error(`Unknown argument ${arg.name.value} found in event directive`);
+        }
+      }
+
+      if (!topic) {
+        throw new Error(`Event directives must have a topic argument`);
+      }
+
+      const configuration = getValueOrDefault(this.eventsConfigurations, this.parentTypeName, () => []);
+      configuration.push({
+        type: eventType,
+        fieldName: this.childName,
+        topic,
+        sourceId,
+      });
+    }
+  }
+
   normalize(document: DocumentNode) {
     const factory = this;
     /* factory.allDirectiveDefinitions is initialized with v1 directive definitions, and v2 definitions are only added
     after the visitor has visited the entire schema and the subgraph is known to be a V2 graph. Consequently,
     allDirectiveDefinitions cannot be used to check for duplicate definitions, and another set (below) is required */
     const definedDirectives = new Set<string>();
-    let isCurrentParentRootType: boolean = false;
     let fieldName = '';
+    const handledRootTypes = new Set<string>();
     // Collect any renamed root types
     visit(document, {
       OperationTypeDefinition: {
@@ -852,6 +948,7 @@ export class NormalizationFactory {
           if (existingOperationType) {
             factory.errors.push(invalidOperationTypeDefinitionError(existingOperationType, newTypeName, operationType));
           } else {
+            handledRootTypes.add(operationType);
             factory.operationTypeNames.set(newTypeName, operationType);
             factory.schemaDefinition.operationTypes.set(operationType, node);
           }
@@ -990,11 +1087,14 @@ export class NormalizationFactory {
       FieldDefinition: {
         enter(node) {
           const name = node.name.value;
-          if (isCurrentParentRootType && (name === SERVICE_FIELD || name === ENTITIES_FIELD)) {
+          if (factory.isCurrentParentRootType && (name === SERVICE_FIELD || name === ENTITIES_FIELD)) {
             return false;
           }
           factory.childName = name;
           factory.lastChildNodeKind = node.kind;
+          if (factory.canContainEventDirectives()) {
+            factory.extractEventDirectives(node);
+          }
           const fieldPath = `${factory.parentTypeName}.${name}`;
           factory.lastChildNodeKind = node.kind;
           const fieldNamedTypeName = getNamedTypeForChild(fieldPath, node.type);
@@ -1213,7 +1313,7 @@ export class NormalizationFactory {
           if (name === SERVICE_OBJECT) {
             return false;
           }
-          isCurrentParentRootType = ROOT_TYPES.has(name) || factory.operationTypeNames.has(name);
+          factory.isCurrentParentRootType = ROOT_TYPES.has(name) || factory.operationTypeNames.has(name);
           factory.parentTypeName = name;
           factory.lastParentNodeKind = node.kind;
           addConcreteTypesForImplementedInterfaces(node, factory.abstractToConcreteTypeNames);
@@ -1244,7 +1344,7 @@ export class NormalizationFactory {
           factory.extractKeyFieldSets(node, fieldSets.keys);
         },
         leave() {
-          isCurrentParentRootType = false;
+          factory.isCurrentParentRootType = false;
           factory.isCurrentParentExtension = false;
           factory.parentTypeName = '';
           factory.lastParentNodeKind = Kind.NULL;
@@ -1256,14 +1356,14 @@ export class NormalizationFactory {
           if (name === SERVICE_OBJECT) {
             return false;
           }
-          isCurrentParentRootType = ROOT_TYPES.has(name) || factory.operationTypeNames.has(name);
+          factory.isCurrentParentRootType = ROOT_TYPES.has(name) || factory.operationTypeNames.has(name);
           factory.parentTypeName = name;
           factory.lastParentNodeKind = node.kind;
           addConcreteTypesForImplementedInterfaces(node, factory.abstractToConcreteTypeNames);
           return factory.handleObjectLikeExtension(node);
         },
         leave() {
-          isCurrentParentRootType = false;
+          factory.isCurrentParentRootType = false;
           factory.isCurrentParentExtension = false;
           factory.parentTypeName = '';
           factory.lastParentNodeKind = Kind.NULL;
@@ -1542,6 +1642,10 @@ export class NormalizationFactory {
             isRootNode: isEntity,
             typeName: parentTypeName,
           };
+          const events = this.eventsConfigurations.get(parentTypeName);
+          if (events) {
+            configurationData.events = events;
+          }
           this.configurationDataMap.set(parentTypeName, configurationData);
           addNonExternalFieldsToSet(parentContainer.fields, configurationData.fieldNames);
           this.validateInterfaceImplementations(parentContainer);

composition/src/normalization/utils.ts

@@ -40,12 +40,14 @@ import {
   INPUT_FIELD_DEFINITION_UPPER,
   INPUT_OBJECT_UPPER,
   INTERFACE_UPPER,
+  MUTATION,
   MUTATION_UPPER,
   OBJECT_UPPER,
   QUERY,
   QUERY_UPPER,
   SCALAR_UPPER,
   SCHEMA_UPPER,
+  SUBSCRIPTION,
   SUBSCRIPTION_UPPER,
   UNION_UPPER,
   VARIABLE_DEFINITION_UPPER,

composition/src/subgraph/field-configuration.ts

@@ -5,12 +5,22 @@ export type RequiredFieldConfiguration = {
   selectionSet: string;
 };
 
+export type EventType = 'subscribe' | 'publish' | 'request';
+
+export type EventConfiguration = {
+  type: EventType;
+  fieldName: string;
+  topic: string;
+  sourceId?: string;
+};
+
 export type ConfigurationData = {
   fieldNames: Set<string>;
   isRootNode: boolean;
   provides?: RequiredFieldConfiguration[];
   keys?: RequiredFieldConfiguration[];
   requires?: RequiredFieldConfiguration[];
+  events?: EventConfiguration[];
   typeName: string;
 };
 

composition/src/utils/constants.ts

@@ -7,6 +7,9 @@ import {
   DEPRECATED,
   ENUM_UPPER,
   ENUM_VALUE_UPPER,
+  EVENTS_PUBLISH,
+  EVENTS_REQUEST,
+  EVENTS_SUBSCRIBE,
   EXTENDS,
   EXTERNAL,
   FIELD_DEFINITION_UPPER,
@@ -26,12 +29,13 @@ import {
   REQUIRES,
   RESOLVABLE,
   SCALAR_UPPER,
-  SCHEMA,
   SCHEMA_UPPER,
   SHAREABLE,
+  SOURCE_ID,
   SPECIFIED_BY,
   STRING_TYPE,
   TAG,
+  TOPIC,
   UNION_UPPER,
 } from './string-constants';
 
@@ -105,6 +109,72 @@ export const BASE_DIRECTIVE_DEFINITIONS: DirectiveDefinitionNode[] = [
     name: stringToNameNode(EXTERNAL),
     repeatable: false,
   },
+  // directive @eventsPublish(topic: String!, sourceID: String) on FIELD_DEFINITION
+  {
+    arguments: [
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(TOPIC),
+        type: {
+          kind: Kind.NON_NULL_TYPE,
+          type: stringToNamedTypeNode(STRING_TYPE),
+        },
+      },
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(SOURCE_ID),
+        type: stringToNamedTypeNode(STRING_TYPE),
+      },
+    ],
+    kind: Kind.DIRECTIVE_DEFINITION,
+    locations: [stringToNameNode(FIELD_DEFINITION_UPPER)],
+    name: stringToNameNode(EVENTS_PUBLISH),
+    repeatable: false,
+  },
+  // directive @eventsRequest(topic: String!, sourceID: String) on FIELD_DEFINITION
+  {
+    arguments: [
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(TOPIC),
+        type: {
+          kind: Kind.NON_NULL_TYPE,
+          type: stringToNamedTypeNode(STRING_TYPE),
+        },
+      },
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(SOURCE_ID),
+        type: stringToNamedTypeNode(STRING_TYPE),
+      },
+    ],
+    kind: Kind.DIRECTIVE_DEFINITION,
+    locations: [stringToNameNode(FIELD_DEFINITION_UPPER)],
+    name: stringToNameNode(EVENTS_REQUEST),
+    repeatable: false,
+  },
+  // directive @eventsSubscribe(topic: String!, sourceID: String) on FIELD_DEFINITION
+  {
+    arguments: [
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(TOPIC),
+        type: {
+          kind: Kind.NON_NULL_TYPE,
+          type: stringToNamedTypeNode(STRING_TYPE),
+        },
+      },
+      {
+        kind: Kind.INPUT_VALUE_DEFINITION,
+        name: stringToNameNode(SOURCE_ID),
+        type: stringToNamedTypeNode(STRING_TYPE),
+      },
+    ],
+    kind: Kind.DIRECTIVE_DEFINITION,
+    locations: [stringToNameNode(FIELD_DEFINITION_UPPER)],
+    name: stringToNameNode(EVENTS_SUBSCRIBE),
+    repeatable: false,
+  },
   // directive @key(fields: openfed__FieldSet!) on INTERFACE | OBJECT
   {
     arguments: [

composition/src/utils/string-constants.ts

@@ -12,6 +12,9 @@ export const ENTITIES_FIELD = '_entities';
 export const ENTITY_UNION = '_Entity';
 export const ENUM_UPPER = 'ENUM';
 export const ENUM_VALUE_UPPER = 'ENUM_VALUE';
+export const EVENTS_PUBLISH = 'eventsPublish';
+export const EVENTS_REQUEST = 'eventsRequest';
+export const EVENTS_SUBSCRIBE = 'eventsSubscribe';
 export const EXTERNAL = 'external';
 export const EXTENDS = 'extends';
 export const EXTENSIONS = 'extensions';
@@ -53,11 +56,13 @@ export const SELECTION_REPRESENTATION = ' { ... }';
 export const SERVICE_OBJECT = '_Service';
 export const SERVICE_FIELD = '_service';
 export const SHAREABLE = 'shareable';
+export const SOURCE_ID = 'sourceID';
 export const SPECIFIED_BY = 'specifiedBy';
 export const STRING_TYPE = 'String';
 export const SUBSCRIPTION = 'Subscription';
 export const SUBSCRIPTION_UPPER = 'SUBSCRIPTION';
 export const TAG = 'tag';
+export const TOPIC = 'topic';
 export const UNION = 'union';
 export const UNION_UPPER = 'UNION';
 export const VARIABLE_DEFINITION_UPPER = 'VARIABLE_DEFINITION';