Restart workflow api (#855)

* restart api

* add missing restartWorkflow

Author: Assem-Uber

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

@@ -0,0 +1,18 @@
+import { type NextRequest } from 'next/server';
+
+import { restartWorkflow } from '@/route-handlers/restart-workflow/restart-workflow';
+import { type RouteParams } from '@/route-handlers/restart-workflow/restart-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: RouteParams }
+) {
+  return routeHandlerWithMiddlewares(
+    restartWorkflow,
+    request,
+    options,
+    routeHandlersDefaultMiddlewares
+  );
+}

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

@@ -0,0 +1,116 @@
+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 { restartWorkflow } from '../restart-workflow';
+import {
+  type RestartWorkflowResponse,
+  type Context,
+} from '../restart-workflow.types';
+
+describe(restartWorkflow.name, () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('calls restartWorkflow and returns valid response', async () => {
+    const { res, mockRestartWorkflow } = await setup({});
+
+    expect(mockRestartWorkflow).toHaveBeenCalledWith({
+      domain: 'mock-domain',
+      workflowExecution: {
+        workflowId: 'mock-wfid',
+        runId: 'mock-runid',
+      },
+      reason: 'Restarting workflow from cadence-web UI',
+    });
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual({ runId: 'mock-runid-new' });
+  });
+
+  it('calls restartWorkflow with custom reason', async () => {
+    const { mockRestartWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        reason: 'This workflow needs to be restarted for various reasons',
+      }),
+    });
+
+    expect(mockRestartWorkflow).toHaveBeenCalledWith(
+      expect.objectContaining({
+        reason: 'This workflow needs to be restarted for various reasons',
+      })
+    );
+  });
+
+  it('returns an error if something went wrong in the backend', async () => {
+    const { res, mockRestartWorkflow } = await setup({
+      error: true,
+    });
+
+    expect(mockRestartWorkflow).toHaveBeenCalled();
+
+    expect(res.status).toEqual(500);
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Could not restart workflow',
+      })
+    );
+  });
+
+  it('returns an error if the request body is not in an expected format', async () => {
+    const { res, mockRestartWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        reason: 5,
+      }),
+    });
+
+    expect(mockRestartWorkflow).not.toHaveBeenCalled();
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Invalid values provided for workflow restart',
+      })
+    );
+  });
+});
+
+async function setup({
+  requestBody,
+  error,
+}: {
+  requestBody?: string;
+  error?: true;
+}) {
+  const mockRestartWorkflow = jest
+    .spyOn(mockGrpcClusterMethods, 'restartWorkflow')
+    .mockImplementationOnce(async () => {
+      if (error) {
+        throw new GRPCError('Could not restart workflow');
+      }
+      return { runId: 'mock-runid-new' } satisfies RestartWorkflowResponse;
+    });
+
+  const res = await restartWorkflow(
+    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, mockRestartWorkflow };
+}

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

@@ -0,0 +1,58 @@
+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 { type Context, type RequestParams } from './restart-workflow.types';
+import restartWorkflowRequestBodySchema from './schemas/restart-workflow-request-body-schema';
+
+export async function restartWorkflow(
+  request: NextRequest,
+  requestParams: RequestParams,
+  ctx: Context
+) {
+  const requestBody = await request.json();
+  const { data, error } =
+    restartWorkflowRequestBodySchema.safeParse(requestBody);
+
+  if (error) {
+    return NextResponse.json(
+      {
+        message: 'Invalid values provided for workflow restart',
+        validationErrors: error.errors,
+      },
+      { status: 400 }
+    );
+  }
+
+  const decodedParams = decodeUrlParams(requestParams.params);
+
+  try {
+    const response = await ctx.grpcClusterMethods.restartWorkflow({
+      domain: decodedParams.domain,
+      workflowExecution: {
+        workflowId: decodedParams.workflowId,
+        runId: decodedParams.runId,
+      },
+      reason: data.reason,
+      // TODO: add user identity
+    });
+
+    return NextResponse.json(response);
+  } catch (e) {
+    logger.error<RouteHandlerErrorPayload>(
+      { requestParams: decodedParams, error: e },
+      'Error restarting workflow'
+    );
+
+    return NextResponse.json(
+      {
+        message:
+          e instanceof GRPCError ? e.message : 'Error restarting workflow',
+        cause: e,
+      },
+      { status: getHTTPStatusCode(e) }
+    );
+  }
+}

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

@@ -0,0 +1,17 @@
+import { type RestartWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/RestartWorkflowExecutionResponse';
+import { type DefaultMiddlewaresContext } from '@/utils/route-handlers-middleware';
+
+export type RouteParams = {
+  domain: string;
+  cluster: string;
+  workflowId: string;
+  runId: string;
+};
+
+export type RequestParams = {
+  params: RouteParams;
+};
+
+export type RestartWorkflowResponse = RestartWorkflowExecutionResponse;
+
+export type Context = DefaultMiddlewaresContext;

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

@@ -0,0 +1,10 @@
+import { z } from 'zod';
+
+const restartWorkflowRequestBodySchema = z.object({
+  reason: z
+    .string()
+    .optional()
+    .default('Restarting workflow from cadence-web UI'),
+});
+
+export default restartWorkflowRequestBodySchema;

src/utils/grpc/grpc-client.ts

@@ -24,6 +24,8 @@ import { type QueryWorkflowRequest__Input } from '@/__generated__/proto-ts/uber/
 import { type QueryWorkflowResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/QueryWorkflowResponse';
 import { type RequestCancelWorkflowExecutionRequest__Input } from '@/__generated__/proto-ts/uber/cadence/api/v1/RequestCancelWorkflowExecutionRequest';
 import { type RequestCancelWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/RequestCancelWorkflowExecutionResponse';
+import { type RestartWorkflowExecutionRequest__Input } from '@/__generated__/proto-ts/uber/cadence/api/v1/RestartWorkflowExecutionRequest';
+import { type RestartWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/RestartWorkflowExecutionResponse';
 import { type SignalWorkflowExecutionRequest__Input } from '@/__generated__/proto-ts/uber/cadence/api/v1/SignalWorkflowExecutionRequest';
 import { type SignalWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/SignalWorkflowExecutionResponse';
 import { type TerminateWorkflowExecutionRequest__Input } from '@/__generated__/proto-ts/uber/cadence/api/v1/TerminateWorkflowExecutionRequest';
@@ -97,6 +99,9 @@ export type GRPCClusterMethods = {
   requestCancelWorkflow: (
     payload: RequestCancelWorkflowExecutionRequest__Input
   ) => Promise<RequestCancelWorkflowExecutionResponse>;
+  restartWorkflow: (
+    payload: RestartWorkflowExecutionRequest__Input
+  ) => Promise<RestartWorkflowExecutionResponse>;
 };
 
 // cache services instances
@@ -283,6 +288,13 @@ const getClusterServicesMethods = async (
       method: 'RequestCancelWorkflowExecution',
       metadata: metadata,
     }),
+    restartWorkflow: workflowService.request<
+      RestartWorkflowExecutionRequest__Input,
+      RestartWorkflowExecutionResponse
+    >({
+      method: 'RestartWorkflowExecution',
+      metadata: metadata,
+    }),
   };
 };
 

src/utils/route-handlers-middleware/middlewares/__mocks__/grpc-cluster-methods.ts

@@ -18,4 +18,5 @@ export const mockGrpcClusterMethods: GRPCClusterMethods = {
   signalWorkflow: jest.fn(),
   terminateWorkflow: jest.fn(),
   requestCancelWorkflow: jest.fn(),
+  restartWorkflow: jest.fn(),
 };