Add configuration for flow concurrency limit [sc-32677] (#388)

Author: hipplej

packages/spectral/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@prismatic-io/spectral",
-  "version": "10.12.2",
+  "version": "10.12.3",
   "description": "Utility library for building Prismatic connectors and code-native integrations",
   "keywords": ["prismatic"],
   "main": "dist/index.js",

packages/spectral/src/cni-testing.test.ts

@@ -260,6 +260,58 @@ describe("test convert flow", () => {
   });
 });
 
+describe("test convert flow with queueConfig", () => {
+  const baseTestFlowInput = {
+    name: "Test Flow",
+    stableKey: "my-test-flow",
+    description: "This is a test flow with queue config",
+    onExecution: async () => {
+      return { data: 123 };
+    },
+  };
+
+  it("passes through valid concurrencyLimit values", () => {
+    const testFlow = flow({
+      ...baseTestFlowInput,
+      queueConfig: {
+        concurrencyLimit: 5,
+      },
+    });
+
+    const result = convertFlow(testFlow, {}, "test-reference-key");
+    expect(result.queueConfig).toMatchObject({
+      usesFifoQueue: false,
+      concurrencyLimit: 5,
+    });
+  });
+
+  it("throws error for concurrencyLimit below minimum (1)", () => {
+    const testFlow = flow({
+      ...baseTestFlowInput,
+      queueConfig: {
+        concurrencyLimit: 1,
+      },
+    });
+
+    expect(() => convertFlow(testFlow, {}, "test-reference-key")).toThrow(
+      "Test Flow has an invalid concurrencyLimit of 1. concurrencyLimit must be between 2 and 10.",
+    );
+  });
+
+  it("throws error for concurrencyLimit above maximum (11)", () => {
+    const testFlow = flow({
+      ...baseTestFlowInput,
+      queueConfig: {
+        concurrencyLimit: 11,
+      },
+    });
+
+    expect(() => convertFlow(testFlow, {}, "test-reference-key")).toThrow(
+      "Test Flow has an invalid concurrencyLimit of 11. concurrencyLimit must be between 2 and 10.",
+    );
+  });
+});
+
 describe("test convert CNI component", () => {
   const myIntegrationDefinition: IntegrationDefinition = {
     name: "My integration",

packages/spectral/src/serverTypes/convertIntegration.ts

@@ -624,6 +624,15 @@ export const convertFlow = (
       );
     }
 
+    if (
+      queueConfig.concurrencyLimit !== undefined &&
+      (queueConfig.concurrencyLimit < 2 || queueConfig.concurrencyLimit > 10)
+    ) {
+      throw new Error(
+        `${flow.name} has an invalid concurrencyLimit of ${queueConfig.concurrencyLimit}. concurrencyLimit must be between 2 and 10.`,
+      );
+    }
+
     result.queueConfig = {
       usesFifoQueue: false, // Should be false by default, even if undefined
       ...queueConfig,

packages/spectral/src/types/IntegrationDefinition.ts

@@ -183,14 +183,16 @@ export type RetryConfig = {
   uniqueRequestIdField?: string;
 };
 
-/** Defines attributes of a retry configuration used by a flow of an integration. */
+/** Defines attributes of a queue configuration used by a flow of an integration. */
 export type QueueConfig = {
   /** Determines whether the flow should be executed using FIFO ordering. Not valid for synchonous or scheduled flows. */
   usesFifoQueue?: boolean;
   /** Reference to the field in the flow's trigger return payload; used to determine whether to queue the execution. */
   dedupeIdField?: string;
   /** Determines whether the flow should be setup for singleton executions. Only valid for scheduled/polling trigger-based flows. */
   singletonExecutions?: boolean;
+  /** The maximum number of concurrent executions for this flow. Must be between 2 and 10. */
+  concurrencyLimit?: number;
 };
 
 /** Defines attributes of a step error configuration used to determine how to handle errors during flow step execution. */