Reset workflow api (#866)

Reset workflow api

Author: Assem-Uber

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

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

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

@@ -0,0 +1,127 @@
+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 { resetWorkflow } from '../reset-workflow';
+import { type Context } from '../reset-workflow.types';
+
+const defaultRequestBody = {
+  reason: 'Resetting workflow from cadence-web UI',
+  decisionFinishEventId: 4,
+  requestId: '',
+  skipSignalReapply: false,
+};
+
+describe(resetWorkflow.name, () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('calls resetWorkflow and returns valid response', async () => {
+    const { res, mockResetWorkflow } = await setup({});
+
+    expect(mockResetWorkflow).toHaveBeenCalledWith({
+      domain: 'mock-domain',
+      workflowExecution: {
+        workflowId: 'mock-wfid',
+        runId: 'mock-runid',
+      },
+      ...defaultRequestBody,
+    });
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual({ runId: 'mock-runid-new' });
+  });
+
+  it('calls resetWorkflow with passed request body', async () => {
+    const { mockResetWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        reason: 'This workflow needs to be reset',
+        decisionFinishEventId: 123,
+        requestId: 'test-request-id',
+        skipSignalReapply: true,
+      }),
+    });
+
+    expect(mockResetWorkflow).toHaveBeenCalledWith(
+      expect.objectContaining({
+        reason: 'This workflow needs to be reset',
+        decisionFinishEventId: 123,
+        requestId: 'test-request-id',
+        skipSignalReapply: true,
+      })
+    );
+  });
+
+  it('returns an error if something went wrong in the backend', async () => {
+    const { res, mockResetWorkflow } = await setup({
+      error: true,
+    });
+
+    expect(mockResetWorkflow).toHaveBeenCalled();
+
+    expect(res.status).toEqual(500);
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Could not reset workflow',
+      })
+    );
+  });
+
+  it('returns an error if the request body is not in an expected format', async () => {
+    const { res, mockResetWorkflow } = await setup({
+      requestBody: JSON.stringify({
+        reason: 5, // should be a string
+        decisionFinishEventId: 'not-a-number', // should be a number
+      }),
+    });
+
+    expect(mockResetWorkflow).not.toHaveBeenCalled();
+
+    const responseJson = await res.json();
+    expect(responseJson).toEqual(
+      expect.objectContaining({
+        message: 'Invalid values provided for workflow reset',
+      })
+    );
+  });
+});
+
+async function setup({
+  requestBody = JSON.stringify(defaultRequestBody),
+  error,
+}: {
+  requestBody?: string;
+  error?: true;
+}) {
+  const mockResetWorkflow = jest
+    .spyOn(mockGrpcClusterMethods, 'resetWorkflow')
+    .mockImplementationOnce(async () => {
+      if (error) {
+        throw new GRPCError('Could not reset workflow');
+      }
+      return { runId: 'mock-runid-new' };
+    });
+
+  const res = await resetWorkflow(
+    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, mockResetWorkflow };
+}

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

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

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

@@ -0,0 +1,17 @@
+import { type ResetWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/ResetWorkflowExecutionResponse';
+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 ResetWorkflowResponse = ResetWorkflowExecutionResponse;
+
+export type Context = DefaultMiddlewaresContext;

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

@@ -0,0 +1,13 @@
+import { z } from 'zod';
+
+const resetWorkflowRequestBodySchema = z.object({
+  reason: z
+    .string()
+    .optional()
+    .default('Resetting workflow from cadence-web UI'),
+  decisionFinishEventId: z.union([z.string(), z.number()]),
+  requestId: z.string().optional(),
+  skipSignalReapply: z.boolean().optional(),
+});
+
+export default resetWorkflowRequestBodySchema;

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 ResetWorkflowExecutionRequest__Input } from '@/__generated__/proto-ts/uber/cadence/api/v1/ResetWorkflowExecutionRequest';
+import { type ResetWorkflowExecutionResponse } from '@/__generated__/proto-ts/uber/cadence/api/v1/ResetWorkflowExecutionResponse';
 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';
@@ -102,6 +104,9 @@ export type GRPCClusterMethods = {
   restartWorkflow: (
     payload: RestartWorkflowExecutionRequest__Input
   ) => Promise<RestartWorkflowExecutionResponse>;
+  resetWorkflow: (
+    payload: ResetWorkflowExecutionRequest__Input
+  ) => Promise<ResetWorkflowExecutionResponse>;
 };
 
 // cache services instances
@@ -295,6 +300,13 @@ const getClusterServicesMethods = async (
       method: 'RestartWorkflowExecution',
       metadata: metadata,
     }),
+    resetWorkflow: workflowService.request<
+      ResetWorkflowExecutionRequest__Input,
+      ResetWorkflowExecutionResponse
+    >({
+      method: 'ResetWorkflowExecution',
+      metadata: metadata,
+    }),
   };
 };
 

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

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