Merge pull request #746 from PROCEED-Labs/persistent-deployment-data

Persistent Executions Data: Initial Commit

Author: Zayn-Javed

src/engine/e2e_tests/process/deployment/deployment.e2e.test.js

@@ -66,7 +66,7 @@ async function startInstance(definitionId, version, engineName, variables = {},
     await request
       .post(`:${engine.port}/process/${definitionId}/versions/${version}/instance`)
       .send({ variables, extras })
-  ).body.instanceId;
+  ).body.processInstanceId;
 }
 
 async function getInstanceInformation(definitionId, instanceId, engineName) {

src/engine/e2e_tests/process/processEndpoint.e2e.test.js

@@ -234,7 +234,7 @@ describe('Test process endpoints', () => {
           expect(getResponse.body).toStrictEqual([]);
           const postResponse = await request.post('/process/definitionId/versions/123/instance');
           expect(postResponse.status).toBe(201);
-          ({ instanceId } = postResponse.body);
+          ({ processInstanceId: instanceId } = postResponse.body);
           // allow everything to start correctly (the user task should have completely started)
           await new Promise((res) => setTimeout(res, 500));
         });

src/engine/universal/core/src/__tests__/management.test.js

@@ -151,8 +151,8 @@ describe('Management', () => {
     jest.spyOn(Engine.prototype, 'deployProcessVersion');
     jest.spyOn(Engine.prototype, 'startProcessVersion');
     distribution.db.isProcessVersionValid.mockResolvedValue(true);
-    const instanceId = await management.createInstance('0', '123', {});
-    expect(management.getEngineWithID(instanceId)).toBeInstanceOf(Engine);
+    const instance = await management.createInstance('0', '123', {});
+    expect(management.getEngineWithID(instance.processInstanceId)).toBeInstanceOf(Engine);
     expect(Engine.prototype.deployProcessVersion).toHaveBeenCalledWith('0', '123', true);
     expect(Engine.prototype.startProcessVersion).toHaveBeenCalledWith(
       '123',
@@ -166,9 +166,11 @@ describe('Management', () => {
   it('reuses an existing ProceedEngine instance when there is one for the given definitionsId to start an instance', async () => {
     jest.spyOn(Engine.prototype, 'deployProcessVersion');
     jest.spyOn(Engine.prototype, 'startProcessVersion');
-    const firstInstanceId = await management.createInstance('0', '123', {});
+    const firstInstance = await management.createInstance('0', '123', {});
+    const firstInstanceId = firstInstance.processInstanceId;
 
-    const secondInstanceId = await management.createInstance('0', '123', {});
+    const secondInstance = await management.createInstance('0', '123', {});
+    const secondInstanceId = secondInstance.processInstanceId;
 
     const firstEngine = management.getEngineWithID(firstInstanceId);
     expect(management.getEngineWithID(firstInstanceId)).toBe(

src/engine/universal/core/src/engine/engine.js

@@ -291,7 +291,7 @@ class Engine {
     this.originalInstanceState = instance;
     this.instanceEventHandlers = {
       onStarted: (newInstance) => {
-        resolver(newInstance.id);
+        resolver(this.getInstanceInformation(newInstance.id));
         // make sure to keep the information from the original instance on the recreated instance
         if (instance && instance.callingInstance) {
           newInstance.callingInstance = instance.callingInstance;

src/engine/universal/core/src/management.js

@@ -510,7 +510,7 @@ const Management = {
       };
     });
 
-    const instanceId = await engine.startProcessVersion(
+    const recoveredInstance = await engine.startProcessVersion(
       processVersion,
       importedInstance.variables,
       importedInstance,
@@ -526,7 +526,7 @@ const Management = {
     // if the instance was in the process of being paused => make sure that it is paused again
     // (will lead to it being paused directly since no tasks have started yet)
     if (importedInstance.instanceState[0] === 'PAUSING') {
-      await engine.pauseInstance(instanceId);
+      await engine.pauseInstance(recoveredInstance.processInstanceId);
     }
 
     // allow waiting instances to be started (and give information about tokens being interrupted which is needed to check if called instances should run)

src/engine/universal/distribution/src/routes/ProcessInstanceRoutes.js

@@ -45,7 +45,7 @@ module.exports = (path, management) => {
       }
     }
 
-    const instanceId = await management.createInstance(
+    const instance = await management.createInstance(
       definitionId,
       version,
       variables,
@@ -57,7 +57,7 @@ module.exports = (path, management) => {
       },
     );
 
-    if (!instanceId) {
+    if (!instance) {
       throw new APIError(
         406,
         `Engine not allowed to start the instance for the process (id: ${definitionId}).`,
@@ -70,7 +70,7 @@ module.exports = (path, management) => {
     return {
       statusCode: 201,
       mimeType: 'application/json',
-      response: JSON.stringify({ instanceId }),
+      response: JSON.stringify(instance),
     };
   });
 

src/management-system-v2/app/(dashboard)/[environmentId]/(automation)/executions/[processId]/process-deployment-view.tsx

@@ -44,6 +44,13 @@ import { useEnvironment } from '@/components/auth-can';
 
 import { GrDocumentUser } from 'react-icons/gr';
 import { handleOpenDocumentation } from '../../../processes/processes-helper';
+import {
+  getProcessStartForm,
+  pauseInstance,
+  resumeInstance,
+  startInstance,
+  stopInstance,
+} from '@/lib/executions/instance-server-actions';
 
 export default function ProcessDeploymentView({
   processId,
@@ -76,15 +83,7 @@ export default function ProcessDeploymentView({
   const canvasRef = useRef<BPMNCanvasRef>(null);
   const [infoPanelOpen, setInfoPanelOpen] = useState(false);
 
-  const {
-    data: deploymentInfo,
-    refetch,
-    startInstance,
-    resumeInstance,
-    pauseInstance,
-    stopInstance,
-    getStartForm,
-  } = useDeployment(processId, initialDeploymentInfo);
+  const { data: deploymentInfo, refetch } = useDeployment(processId, initialDeploymentInfo);
 
   const {
     selectedVersion,
@@ -341,7 +340,7 @@ export default function ProcessDeploymentView({
                           const latestDeployment = getLatestDeployment(deploymentInfo);
                           const versionId = latestDeployment.versionId;
 
-                          let startForm = await getStartForm(versionId);
+                          let startForm = await getProcessStartForm(spaceId, processId, versionId);
 
                           if (typeof startForm !== 'string') return startForm;
 
@@ -370,7 +369,7 @@ export default function ProcessDeploymentView({
 
                             setStartForm(startForm);
                           } else {
-                            return startInstance(versionId);
+                            return startInstance(spaceId, processId, versionId);
                           }
                         },
                         onSuccess: async (instanceId) => {
@@ -442,7 +441,12 @@ export default function ProcessDeploymentView({
                         onClick={async () => {
                           setResumingInstance(true);
                           await wrapServerCall({
-                            fn: () => resumeInstance(selectedInstance.processInstanceId),
+                            fn: () =>
+                              resumeInstance(
+                                spaceId,
+                                processId,
+                                selectedInstance.processInstanceId,
+                              ),
                             onSuccess: async () => await refetch(),
                           });
                           setResumingInstance(false);
@@ -460,7 +464,8 @@ export default function ProcessDeploymentView({
                         onClick={async () => {
                           setPausingInstance(true);
                           await wrapServerCall({
-                            fn: async () => pauseInstance(selectedInstance.processInstanceId),
+                            fn: async () =>
+                              pauseInstance(spaceId, processId, selectedInstance.processInstanceId),
                             onSuccess: async () => await refetch(),
                           });
                           setPausingInstance(false);
@@ -478,7 +483,8 @@ export default function ProcessDeploymentView({
                       onClick={async () => {
                         setStoppingInstance(true);
                         await wrapServerCall({
-                          fn: async () => stopInstance(selectedInstance.processInstanceId),
+                          fn: async () =>
+                            stopInstance(spaceId, processId, selectedInstance.processInstanceId),
                           onSuccess: async () => await refetch(),
                         });
                         setStoppingInstance(false);
@@ -547,7 +553,7 @@ export default function ProcessDeploymentView({
 
             // start the instance with the initial variable values from the start form
             await wrapServerCall({
-              fn: () => startInstance(versionId, mappedVariables),
+              fn: () => startInstance(spaceId, processId, versionId, mappedVariables),
 
               onSuccess: async (instanceId) => {
                 await refetch();

src/management-system-v2/app/(dashboard)/[environmentId]/(automation)/executions/deployment-hook.ts

@@ -1,21 +1,12 @@
-import { useEnvironment, useSession } from '@/components/auth-can';
+import { useEnvironment } from '@/components/auth-can';
 import {
   DeployedProcessInfo,
   InstanceInfo,
   VersionInfo,
   getDeployments,
 } from '@/lib/engines/deployment';
-import {
-  pauseInstanceOnMachine,
-  resumeInstanceOnMachine,
-  startInstanceOnMachine,
-  stopInstanceOnMachine,
-} from '@/lib/engines/instances';
-import { Engine } from '@/lib/engines/types';
-import { getStartFormFromMachine } from '@/lib/engines/tasklist';
 import useEngines from '@/lib/engines/use-engines';
-import { asyncFilter, asyncForEach, deepEquals } from '@/lib/helpers/javascriptHelpers';
-import { getErrorMessage, userError } from '@/lib/user-error';
+import { deepEquals } from '@/lib/helpers/javascriptHelpers';
 import { useQuery } from '@tanstack/react-query';
 import { useCallback } from 'react';
 
@@ -89,7 +80,6 @@ const mergeDeployment = (
 
 function useDeployment(definitionId: string, initialData?: DeployedProcessInfo) {
   const space = useEnvironment();
-  const { data: session } = useSession();
 
   const { data: engines } = useEngines(space, {
     key: [definitionId],
@@ -99,90 +89,6 @@ function useDeployment(definitionId: string, initialData?: DeployedProcessInfo)
     },
   });
 
-  const startInstance = async (versionId: string, variables: { [key: string]: any } = {}) => {
-    if (!engines?.length) return userError('No fitting engine found');
-
-    // TODO: in case of static deployment or different versions on different engines we will have
-    // to check if the engine can actually be used to start an instance
-    return await startInstanceOnMachine(definitionId, versionId, engines[0], variables, {
-      processInitiator: session?.user.id,
-      spaceIdOfProcessInitiator: space.spaceId,
-    });
-  };
-
-  const activeStates = ['PAUSED', 'RUNNING', 'READY', 'DEPLOYMENT-WAITING', 'WAITING'];
-  async function changeInstanceState(
-    instanceId: string,
-    stateValidator: (state: InstanceInfo['instanceState']) => boolean,
-    stateChangeFunction: typeof resumeInstanceOnMachine,
-  ) {
-    if (!engines) return;
-    try {
-      const targetEngines = await asyncFilter(engines, async (engine: Engine) => {
-        const deployments = await getDeployments([engine]);
-
-        return deployments.some((deployment) => {
-          if (deployment.definitionId !== definitionId) return false;
-
-          const instance = deployment.instances.find(
-            (instance) => instance.processInstanceId === instanceId,
-          );
-          if (!instance) return false;
-
-          return stateValidator(instance.instanceState);
-        });
-      });
-
-      await asyncForEach(targetEngines, async (engine) => {
-        await stateChangeFunction(definitionId, instanceId, engine);
-      });
-    } catch (e) {
-      const message = getErrorMessage(e);
-      return userError(message);
-    }
-  }
-
-  async function resumeInstance(instanceId: string) {
-    // TODO: manage permissions for starting an instance
-    return await changeInstanceState(
-      instanceId,
-      (tokenStates) => tokenStates.some((tokenState) => tokenState === 'PAUSED'),
-      resumeInstanceOnMachine,
-    );
-  }
-
-  async function pauseInstance(instanceId: string) {
-    // TODO: manage permissions for starting an instance
-    return await changeInstanceState(
-      instanceId,
-      (tokenStates) =>
-        tokenStates.some((state) => activeStates.includes(state) && state !== 'PAUSED'),
-      pauseInstanceOnMachine,
-    );
-  }
-
-  async function stopInstance(instanceId: string) {
-    // TODO: manage permissions for starting an instance
-    return await changeInstanceState(
-      instanceId,
-      (tokenStates) => tokenStates.some((state) => activeStates.includes(state)),
-      stopInstanceOnMachine,
-    );
-  }
-
-  async function getStartForm(versionId: string) {
-    if (!engines?.length) return userError('No fitting engine found');
-
-    try {
-      // TODO: in case of static deployment or different versions on different engines we will have
-      // to check if the engine can actually be used to start an instance
-      return await getStartFormFromMachine(definitionId, versionId, engines[0]);
-    } catch (e) {
-      const message = getErrorMessage(e);
-      return userError(message);
-    }
-  }
-
   const queryFn = useCallback(async () => {
     if (engines?.length) {
       // TODO: this only handles situations where we have only a single engine
@@ -212,7 +118,7 @@ function useDeployment(definitionId: string, initialData?: DeployedProcessInfo)
     },
   });
 
-  return { ...query, startInstance, resumeInstance, pauseInstance, stopInstance, getStartForm };
+  return { ...query };
 }
 
 export default useDeployment;

src/management-system-v2/lib/data/deployment.ts

@@ -0,0 +1,54 @@
+'use server';
+
+import db from '@/lib/data/db';
+import { DeploymentInput, DeploymentInputSchema } from '../deployment-schema';
+import { getCurrentEnvironment } from '@/components/auth';
+import { UserErrorType, userError } from '../user-error';
+
+export async function getProcessDeployments(spaceId: string, processId: string) {
+  const { ability } = await getCurrentEnvironment(spaceId);
+
+  if (!ability.can('view', 'Execution'))
+    return userError('Invalid Permissions', UserErrorType.PermissionError);
+
+  const deployments = await db.processDeployment.findMany({
+    where: { AND: [{ version: { processId } }, { removeTime: null }] },
+    include: { version: { select: { processId: true } } },
+  });
+
+  return deployments.map((d) => ({ ...d, version: undefined, processId: d.version.processId }));
+}
+
+export async function addDeployment(spaceId: string, input: DeploymentInput) {
+  const { ability } = await getCurrentEnvironment(spaceId);
+
+  if (!ability.can('create', 'Execution'))
+    return userError('Invalid Permissions', UserErrorType.PermissionError);
+
+  const data = DeploymentInputSchema.parse(input);
+
+  await db.processDeployment.createMany({
+    data: data.engineIds.map((engineId) => ({ ...data, engineIds: undefined, engineId })),
+  });
+}
+
+export async function updateDeployment(
+  spaceId: string,
+  deploymentId: string,
+  input: Partial<DeploymentInput>,
+) {
+  const { ability } = await getCurrentEnvironment(spaceId);
+
+  if (!ability.can('update', 'Execution')) {
+    return userError('Invalid Permissions', UserErrorType.PermissionError);
+  }
+
+  const data = DeploymentInputSchema.partial().strict().parse(input);
+
+  const result = await db.processDeployment.update({
+    where: { id: deploymentId },
+    data,
+  });
+
+  return result;
+}