Signal workflow API (#902)

Author: Assem-Uber

src/app/api/domains/[domain]/[cluster]/workflows/[workflowId]/[runId]/(actions)/signal/route.ts

@@ -0,0 +1,18 @@
+import { type NextRequest } from 'next/server';
+
+import { signalWorkflow } from '@/route-handlers/signal-workflow/signal-workflow';
+import { type RequestParams } from '@/route-handlers/signal-workflow/signal-workflow.types';
+import { routeHandlerWithMiddlewares } from '@/utils/route-handlers-middleware';
+import routeHandlersDefaultMiddlewares from '@/utils/route-handlers-middleware/config/route-handlers-default-middlewares.config';
+
+export async function POST(
+  request: NextRequest,
+  options: { params: RequestParams['params'] }
+) {
+  return routeHandlerWithMiddlewares(
+    signalWorkflow,
+    request,
+    options,
+    routeHandlersDefaultMiddlewares
+  );
+}

src/route-handlers/signal-workflow/__tests__/signal-workflow.node.ts

@@ -0,0 +1,124 @@
+import { NextRequest } from 'next/server';
+
+import { GRPCError } from '@/utils/grpc/grpc-error';
+import { mockGrpcClusterMethods } from '@/utils/route-handlers-middleware/middlewares/__mocks__/grpc-cluster-methods';
+
+import { signalWorkflow } from '../signal-workflow';
+import {
+  type SignalWorkflowRequestBody,
+  type Context,
+} from '../signal-workflow.types';
+
+const defaultRequestBody = {
+  signalName: 'test-signal',
+  signalInput: '"test-input"',
+} satisfies SignalWorkflowRequestBody;
+
+describe(signalWorkflow.name, () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('calls signalWorkflow and returns valid response', async () => {
+    const { res, mockSignalWorkflow } = await setup({});
+
+    expect(mockSignalWorkflow).toHaveBeenCalledWith({
+      domain: 'mock-domain',
+      workflowExecution: {
+        workflowId: 'mock-wfid',
+        runId: 'mock-runid',
+      },
+      signalName: 'test-signal',
+      signalInput: { data: Buffer.from('"test-input"') },
+    });
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual({});
+  });
+
+  it('calls signalWorkflow without signalInput when not provided', async () => {
+    const { mockSignalWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        signalName: 'signal-without-input',
+      }),
+    });
+
+    expect(mockSignalWorkflow).toHaveBeenCalledWith(
+      expect.objectContaining({
+        signalName: 'signal-without-input',
+        signalInput: undefined,
+      })
+    );
+  });
+
+  it('returns an error if something went wrong in the backend', async () => {
+    const { res, mockSignalWorkflow } = await setup({
+      error: true,
+    });
+
+    expect(mockSignalWorkflow).toHaveBeenCalled();
+
+    expect(res.status).toEqual(500);
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Could not signal workflow',
+      })
+    );
+  });
+
+  it('returns an error if the signal input has an unexpected format', async () => {
+    const { res, mockSignalWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        signalName: 'test-signal',
+        signalInput: 'not-an-object', // should be an object
+      } satisfies SignalWorkflowRequestBody),
+    });
+
+    expect(mockSignalWorkflow).not.toHaveBeenCalled();
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Invalid values provided for workflow signal',
+      })
+    );
+  });
+});
+
+async function setup({
+  requestBody = JSON.stringify(defaultRequestBody),
+  error,
+}: {
+  requestBody?: string;
+  error?: true;
+}) {
+  const mockSignalWorkflow = jest
+    .spyOn(mockGrpcClusterMethods, 'signalWorkflow')
+    .mockImplementationOnce(async () => {
+      if (error) {
+        throw new GRPCError('Could not signal workflow');
+      }
+      return {};
+    });
+
+  const res = await signalWorkflow(
+    new NextRequest('http://localhost', {
+      method: 'POST',
+      body: requestBody ?? '{}',
+    }),
+    {
+      params: {
+        domain: 'mock-domain',
+        cluster: 'mock-cluster',
+        workflowId: 'mock-wfid',
+        runId: 'mock-runid',
+      },
+    },
+    {
+      grpcClusterMethods: mockGrpcClusterMethods,
+    } as Context
+  );
+
+  return { res, mockSignalWorkflow };
+}

src/route-handlers/signal-workflow/schemas/signal-workflow-input-schema.ts

@@ -0,0 +1,16 @@
+import { z } from 'zod';
+
+import losslessJsonParse from '@/utils/lossless-json-parse';
+
+const signalWorkflowInputSchema = z.string().superRefine((str, ctx) => {
+  if (!str) return undefined;
+
+  try {
+    return losslessJsonParse(str);
+  } catch {
+    ctx.addIssue({ code: 'custom', message: 'Invalid JSON' });
+    return z.NEVER;
+  }
+});
+
+export default signalWorkflowInputSchema;

src/route-handlers/signal-workflow/schemas/signal-workflow-request-body-schema.ts

@@ -0,0 +1,10 @@
+import { z } from 'zod';
+
+import signalWorkflowInputSchema from './signal-workflow-input-schema';
+
+const signalWorkflowRequestBodySchema = z.object({
+  signalName: z.string().min(1),
+  signalInput: signalWorkflowInputSchema.optional(),
+});
+
+export default signalWorkflowRequestBodySchema;

src/route-handlers/signal-workflow/signal-workflow.ts

@@ -0,0 +1,61 @@
+import { type NextRequest, NextResponse } from 'next/server';
+
+import decodeUrlParams from '@/utils/decode-url-params';
+import { getHTTPStatusCode, GRPCError } from '@/utils/grpc/grpc-error';
+import logger, { type RouteHandlerErrorPayload } from '@/utils/logger';
+
+import signalWorkflowRequestBodySchema from './schemas/signal-workflow-request-body-schema';
+import { type Context, type RequestParams } from './signal-workflow.types';
+
+export async function signalWorkflow(
+  request: NextRequest,
+  requestParams: RequestParams,
+  ctx: Context
+) {
+  const requestBody = await request.json();
+  const { data, error } =
+    signalWorkflowRequestBodySchema.safeParse(requestBody);
+
+  if (error) {
+    return NextResponse.json(
+      {
+        message: 'Invalid values provided for workflow signal',
+        validationErrors: error.errors,
+      },
+      { status: 400 }
+    );
+  }
+
+  const decodedParams = decodeUrlParams(requestParams.params);
+
+  try {
+    const response = await ctx.grpcClusterMethods.signalWorkflow({
+      domain: decodedParams.domain,
+      workflowExecution: {
+        workflowId: decodedParams.workflowId,
+        runId: decodedParams.runId,
+      },
+      signalName: data.signalName,
+      signalInput: data.signalInput
+        ? { data: Buffer.from(data.signalInput) }
+        : undefined,
+      // TODO: add user identity
+    });
+
+    return NextResponse.json(response);
+  } catch (e) {
+    logger.error<RouteHandlerErrorPayload>(
+      { requestParams: decodedParams, error: e },
+      'Error signaling workflow'
+    );
+
+    return NextResponse.json(
+      {
+        message:
+          e instanceof GRPCError ? e.message : 'Error signaling workflow',
+        cause: e,
+      },
+      { status: getHTTPStatusCode(e) }
+    );
+  }
+}

src/route-handlers/signal-workflow/signal-workflow.types.ts

@@ -0,0 +1,22 @@
+import { type z } from 'zod';
+
+import { type GRPCClusterMethods } from '@/utils/grpc/grpc-client';
+
+import type signalWorkflowRequestBodySchema from './schemas/signal-workflow-request-body-schema';
+
+export type RequestParams = {
+  params: {
+    domain: string;
+    cluster: string;
+    workflowId: string;
+    runId: string;
+  };
+};
+
+export type SignalWorkflowRequestBody = z.infer<
+  typeof signalWorkflowRequestBodySchema
+>;
+
+export type Context = {
+  grpcClusterMethods: GRPCClusterMethods;
+};