feat: add redis pubsub support to EDFS

Makefile

@@ -12,6 +12,7 @@ setup-dev-tools: setup-build-tools
 	go install github.com/yannh/kubeconform/cmd/kubeconform@v0.6.3
 	go install github.com/norwoodj/helm-docs/cmd/helm-docs@v1.11.3
 	go install github.com/vektra/mockery/v3@v3.3.1
+	go install github.com/Antonboom/testifylint@v1.6.1
 
 prerequisites: setup-dev-tools
 	go version
@@ -20,6 +21,7 @@ prerequisites: setup-dev-tools
 	docker -v
 	dbmate -v
 	mockery version
+	testifylint -V=full
 
 infra-up: dc-dev
 

composition/src/router-configuration/types.ts

@@ -2,6 +2,8 @@ export type NatsEventType = 'subscribe' | 'publish' | 'request';
 
 export type KafkaEventType = 'subscribe' | 'publish';
 
+export type RedisEventType = 'subscribe' | 'publish';
+
 export type StreamConfiguration = {
   consumerInactiveThreshold: number;
   consumerName: string;
@@ -25,7 +27,15 @@ export type NatsEventConfiguration = {
   streamConfiguration?: StreamConfiguration;
 };
 
-export type EventConfiguration = KafkaEventConfiguration | NatsEventConfiguration;
+export type RedisEventConfiguration = {
+  fieldName: string;
+  providerId: string;
+  providerType: 'redis';
+  channels: string[];
+  type: RedisEventType;
+};
+
+export type EventConfiguration = KafkaEventConfiguration | NatsEventConfiguration | RedisEventConfiguration;
 
 export type SubscriptionFilterValue = boolean | null | number | string;
 

composition/src/utils/string-constants.ts

@@ -29,6 +29,8 @@ export const EDFS_NATS_REQUEST = 'edfs__natsRequest';
 export const EDFS_NATS_SUBSCRIBE = 'edfs__natsSubscribe';
 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 ENTITIES = 'entities';
 export const ENTITIES_FIELD = '_entities';
 export const ENTITY_UNION = '_Entity';
@@ -83,6 +85,7 @@ export const MUTATION_UPPER = 'MUTATION';
 export const PROPAGATE = 'propagate';
 export const PROVIDER_TYPE_KAFKA = 'kafka';
 export const PROVIDER_TYPE_NATS = 'nats';
+export const PROVIDER_TYPE_REDIS = 'redis';
 export const NOT_APPLICABLE = 'N/A';
 export const NAME = 'name';
 export const NON_NULLABLE_EDFS_PUBLISH_EVENT_RESULT = 'edfs__PublishResult!';
@@ -146,6 +149,8 @@ export const UNION_UPPER = 'UNION';
 export const URL_LOWER = 'url';
 export const VALUES = 'values';
 export const VARIABLE_DEFINITION_UPPER = 'VARIABLE_DEFINITION';
+export const CHANNEL = 'channel';
+export const CHANNELS = 'channels';
 
 export const EXECUTABLE_DIRECTIVE_LOCATIONS = new Set<string>([
   FIELD_UPPER,

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

@@ -10,6 +10,8 @@ import {
   EDFS_NATS_PUBLISH_DEFINITION,
   EDFS_NATS_REQUEST_DEFINITION,
   EDFS_NATS_SUBSCRIBE_DEFINITION,
+  EDFS_REDIS_PUBLISH_DEFINITION,
+  EDFS_REDIS_SUBSCRIBE_DEFINITION,
   EXTENDS_DEFINITION,
   EXTERNAL_DEFINITION,
   INACCESSIBLE_DEFINITION,
@@ -92,6 +94,10 @@ import {
   TOPICS,
   UNION_UPPER,
   URL_LOWER,
+  CHANNEL,
+  CHANNELS,
+  EDFS_REDIS_PUBLISH,
+  EDFS_REDIS_SUBSCRIBE,
 } from '../../utils/string-constants';
 
 export const AUTHENTICATED_DEFINITION_DATA: DirectiveDefinitionData = {
@@ -433,6 +439,70 @@ export const NATS_SUBSCRIBE_DEFINITION_DATA: DirectiveDefinitionData = {
   requiredArgumentNames: new Set<string>([SUBJECTS]),
 };
 
+export const REDIS_PUBLISH_DEFINITION_DATA: DirectiveDefinitionData = {
+  argumentTypeNodeByArgumentName: new Map<string, ArgumentData>([
+    [
+      CHANNEL,
+      {
+        name: CHANNEL,
+        typeNode: REQUIRED_STRING_TYPE_NODE,
+      },
+    ],
+    [
+      PROVIDER_ID,
+      {
+        name: PROVIDER_ID,
+        typeNode: REQUIRED_STRING_TYPE_NODE,
+        defaultValue: {
+          kind: Kind.STRING,
+          value: DEFAULT_EDFS_PROVIDER_ID,
+        },
+      },
+    ],
+  ]),
+  isRepeatable: false,
+  locations: new Set<string>([FIELD_DEFINITION_UPPER]),
+  name: EDFS_REDIS_PUBLISH,
+  node: EDFS_REDIS_PUBLISH_DEFINITION,
+  optionalArgumentNames: new Set<string>([PROVIDER_ID]),
+  requiredArgumentNames: new Set<string>([CHANNEL]),
+};
+
+export const REDIS_SUBSCRIBE_DEFINITION_DATA: DirectiveDefinitionData = {
+  argumentTypeNodeByArgumentName: new Map<string, ArgumentData>([
+    [
+      CHANNELS,
+      {
+        name: CHANNELS,
+        typeNode: {
+          kind: Kind.NON_NULL_TYPE,
+          type: {
+            kind: Kind.LIST_TYPE,
+            type: REQUIRED_STRING_TYPE_NODE,
+          },
+        },
+      },
+    ],
+    [
+      PROVIDER_ID,
+      {
+        name: PROVIDER_ID,
+        typeNode: REQUIRED_STRING_TYPE_NODE,
+        defaultValue: {
+          kind: Kind.STRING,
+          value: DEFAULT_EDFS_PROVIDER_ID,
+        },
+      },
+    ],
+  ]),
+  isRepeatable: false,
+  locations: new Set<string>([FIELD_DEFINITION_UPPER]),
+  name: EDFS_REDIS_SUBSCRIBE,
+  node: EDFS_REDIS_SUBSCRIBE_DEFINITION,
+  optionalArgumentNames: new Set<string>([PROVIDER_ID]),
+  requiredArgumentNames: new Set<string>([CHANNELS]),
+};
+
 export const OVERRIDE_DEFINITION_DATA: DirectiveDefinitionData = {
   argumentTypeNodeByArgumentName: new Map<string, ArgumentData>([
     [

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

@@ -326,6 +326,11 @@ import {
   SUCCESS,
   TOPIC,
   TOPICS,
+  CHANNEL,
+  CHANNELS,
+  EDFS_REDIS_PUBLISH,
+  EDFS_REDIS_SUBSCRIBE,
+  PROVIDER_TYPE_REDIS,
 } from '../../utils/string-constants';
 import { MAX_INT32 } from '../../utils/integer-constants';
 import {
@@ -2413,6 +2418,89 @@ export class NormalizationFactory {
     };
   }
 
+  getRedisPublishConfiguration(
+    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 CHANNEL: {
+          if (argumentNode.value.kind !== Kind.STRING || argumentNode.value.value.length < 1) {
+            errorMessages.push(invalidEventSubjectErrorMessage(CHANNEL));
+            continue;
+          }
+          validateArgumentTemplateReferences(argumentNode.value.value, argumentDataByArgumentName, errorMessages);
+          channels.push(argumentNode.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_REDIS, channels, type: PUBLISH };
+  }
+
+  getRedisSubscribeConfiguration(
+    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_REDIS,
+      channels,
+      type: SUBSCRIBE,
+    };
+  }
+
   validateSubscriptionFilterDirectiveLocation(node: FieldDefinitionNode) {
     if (!node.directives) {
       return;
@@ -2490,6 +2578,24 @@ export class NormalizationFactory {
           );
           break;
         }
+        case EDFS_REDIS_PUBLISH: {
+          eventConfiguration = this.getRedisPublishConfiguration(
+            directive,
+            argumentDataByArgumentName,
+            fieldName,
+            errorMessages,
+          );
+          break;
+        }
+        case EDFS_REDIS_SUBSCRIBE: {
+          eventConfiguration = this.getRedisSubscribeConfiguration(
+            directive,
+            argumentDataByArgumentName,
+            fieldName,
+            errorMessages,
+          );
+          break;
+        }
         default:
           continue;
       }
@@ -2515,11 +2621,11 @@ export class NormalizationFactory {
   getValidEventsDirectiveNamesForOperationTypeNode(operationTypeNode: OperationTypeNode): Set<string> {
     switch (operationTypeNode) {
       case OperationTypeNode.MUTATION:
-        return new Set<string>([EDFS_KAFKA_PUBLISH, EDFS_NATS_PUBLISH, EDFS_NATS_REQUEST]);
+        return new Set<string>([EDFS_KAFKA_PUBLISH, EDFS_NATS_PUBLISH, EDFS_NATS_REQUEST, EDFS_REDIS_PUBLISH]);
       case OperationTypeNode.QUERY:
         return new Set<string>([EDFS_NATS_REQUEST]);
       case OperationTypeNode.SUBSCRIPTION:
-        return new Set<string>([EDFS_KAFKA_SUBSCRIBE, EDFS_NATS_SUBSCRIBE]);
+        return new Set<string>([EDFS_KAFKA_SUBSCRIBE, EDFS_NATS_SUBSCRIBE, EDFS_REDIS_SUBSCRIBE]);
     }
   }
 

composition/src/v1/normalization/utils.ts

@@ -37,6 +37,8 @@ import {
   NATS_PUBLISH_DEFINITION_DATA,
   NATS_REQUEST_DEFINITION_DATA,
   NATS_SUBSCRIBE_DEFINITION_DATA,
+  REDIS_PUBLISH_DEFINITION_DATA,
+  REDIS_SUBSCRIBE_DEFINITION_DATA,
   OVERRIDE_DEFINITION_DATA,
   PROVIDES_DEFINITION_DATA,
   REQUIRES_DEFINITION_DATA,
@@ -57,6 +59,8 @@ import {
   EDFS_NATS_PUBLISH,
   EDFS_NATS_REQUEST,
   EDFS_NATS_SUBSCRIBE,
+  EDFS_REDIS_PUBLISH,
+  EDFS_REDIS_SUBSCRIBE,
   EXTENDS,
   EXTERNAL,
   FIELDS,
@@ -390,6 +394,8 @@ export function initializeDirectiveDefinitionDatas(): Map<string, DirectiveDefin
     [EDFS_NATS_PUBLISH, NATS_PUBLISH_DEFINITION_DATA],
     [EDFS_NATS_REQUEST, NATS_REQUEST_DEFINITION_DATA],
     [EDFS_NATS_SUBSCRIBE, NATS_SUBSCRIBE_DEFINITION_DATA],
+    [EDFS_REDIS_PUBLISH, REDIS_PUBLISH_DEFINITION_DATA],
+    [EDFS_REDIS_SUBSCRIBE, REDIS_SUBSCRIBE_DEFINITION_DATA],
     [EXTENDS, EXTENDS_DEFINITION_DATA],
     [EXTERNAL, EXTERNAL_DEFINITION_DATA],
     [INACCESSIBLE, INACCESSIBLE_DEFINITION_DATA],