Optionally predict next states

Author: jasny

package.json

@@ -20,7 +20,7 @@
     "test:e2e": "jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
-    "@letsflow/core": "^1.0.3",
+    "@letsflow/core": "^1.0.4",
     "@nestjs/common": "^10.3.9",
     "@nestjs/core": "^10.3.9",
     "@nestjs/event-emitter": "^3.0.0",
@@ -29,6 +29,7 @@
     "@nestjs/swagger": "^7.3.1",
     "ajv": "^8.16.0",
     "amqp-connection-manager": "^4.1.14",
+    "amqplib": "^0.10.5",
     "buffer-crc32": "^1.0.0",
     "convict": "^6.2.4",
     "mongodb": "^6.7.0",

src/config/test.json

@@ -1,7 +1,13 @@
 {
   "db": "mongodb://localhost:27017/letsflow-test",
-  "summeryFields": {
-    "scenario": ["extraScenarioField"],
-    "process": ["extraProcessField"]
+  "scenario": {
+    "summeryFields": [
+      "extraScenarioField"
+    ]
+  },
+  "process": {
+    "summeryFields": [
+      "extraProcessField"
+    ]
   }
 }

src/process/process.controller.ts

@@ -8,13 +8,14 @@ import {
   ApiOperation,
   ApiParam,
   ApiProduces,
+  ApiQuery,
   ApiResponse,
   ApiTags,
 } from '@nestjs/swagger';
 import { StartInstructions } from './process.dto';
 import { Response } from 'express';
 import { Account, ApiKey, ApiPrivilege, AuthApiKey, AuthGuard, AuthUser } from '@/auth';
-import { ActionEvent, Process } from '@letsflow/core/process';
+import { ActionEvent, instantiate, Process } from '@letsflow/core/process';
 import { AuthService } from '@/auth/auth.service';
 import { ScenarioService } from '@/scenario/scenario.service';
 import { Scenario } from '@letsflow/core/scenario';
@@ -91,21 +92,23 @@ export class ProcessController {
 
   @ApiOperation({ summary: 'Get a process by ID' })
   @ApiParam({ name: 'id', description: 'Process ID', format: 'uuid' })
+  @ApiQuery({ name: 'predict', description: 'Predict next states', type: 'boolean' })
   @ApiResponse({ status: 200, description: 'Success' })
   @ApiProduces('application/json')
   @Get('/:id')
   public async get(
     @Param('id') id: string,
     @AuthUser() user: Account | undefined,
     @AuthApiKey() apiKey: ApiKey | undefined,
+    @Query('predict') addPrediction: boolean | undefined,
     @Res() res: Response,
   ): Promise<void> {
     if (!(await this.processes.has(id))) {
       res.status(404).send('Process not found');
       return;
     }
 
-    const process = await this.processes.get(id);
+    let process = await this.processes.get(id);
 
     if (apiKey && !this.isAllowedByApiKey(apiKey, process.scenario)) {
       res.status(403).send('Insufficient privileges of API key');
@@ -117,17 +120,22 @@ export class ProcessController {
       return;
     }
 
+    if (addPrediction) {
+      process = this.processes.predict(process);
+    }
+
     res.status(200).json(process);
   }
 
   @ApiOperation({ summary: 'Start a process' })
-  @ApiConsumes('application/json')
+  @ApiQuery({ name: 'predict', description: 'Predict next states', type: 'boolean' })
   @ApiHeader({
     name: 'As-Actor',
     description:
       'Specify actor when multiple actors could have performed the action and actor cannot be determined based on the user',
     required: false,
   })
+  @ApiConsumes('application/json')
   @ApiBody({ required: true, type: StartInstructions })
   @ApiResponse({ status: 201, description: 'Created' })
   @ApiPrivilege('process:start')
@@ -136,6 +144,7 @@ export class ProcessController {
     @Headers('As-Actor') actor: string | undefined,
     @AuthUser() user: Account | undefined,
     @AuthApiKey() apiKey: ApiKey | undefined,
+    @Query('predict') addPrediction: boolean | undefined,
     @Body() instructions: StartInstructions,
     @Res() res: Response,
   ): Promise<void> {
@@ -153,16 +162,19 @@ export class ProcessController {
       return;
     }
 
-    let process = await this.processes.instantiate(scenario);
+    let process = instantiate(scenario);
 
     const { key: action, response } =
       typeof instructions.action === 'object' ? instructions.action : { key: instructions.action, response: undefined };
 
     process = await this.doStep(process, user, action, actor, response, res);
+    if (!process) return;
 
-    if (process) {
-      res.status(201).header('Location', `/processes/${process.id}`).json(process);
+    if (addPrediction) {
+      process = this.processes.predict(process);
     }
+
+    res.status(201).header('Location', `/processes/${process.id}`).json(process);
   }
 
   @ApiOperation({ summary: 'Step through a process' })
@@ -174,8 +186,9 @@ export class ProcessController {
       'Specify actor when multiple actors could have performed the action and actor cannot be determined based on the user',
     required: false,
   })
+  @ApiConsumes('application/json')
   @ApiBody({ required: true })
-  @ApiResponse({ status: 204, description: 'No Content' })
+  @ApiResponse({ status: 200, description: 'Success' })
   @ApiPrivilege('process:step')
   @Post(':id/:action')
   public async step(
@@ -184,6 +197,7 @@ export class ProcessController {
     @Headers('As-Actor') actor: string | undefined,
     @AuthUser() user: Account | undefined,
     @AuthApiKey() apiKey: ApiKey | undefined,
+    @Query('predict') addPrediction: boolean | undefined,
     @Body() body: any,
     @Res() res: Response,
   ): Promise<void> {
@@ -205,10 +219,13 @@ export class ProcessController {
     }
 
     process = await this.doStep(process, user, action, actor, body, res);
+    if (!process) return;
 
-    if (process) {
-      res.status(200).json(process);
+    if (addPrediction) {
+      process = this.processes.predict(process);
     }
+
+    res.status(200).json(process);
   }
 
   private async doStep(

src/process/process.service.spec.ts

@@ -7,17 +7,28 @@ import { ScenarioService } from '@/scenario/scenario.service';
 import { instantiate, InstantiateEvent, step } from '@letsflow/core/process';
 import { normalize } from '@letsflow/core/scenario';
 import { uuid } from '@letsflow/core';
-import { from as bsonUUID } from 'uuid-mongodb';
+import { from as bsonUUID, MUUID } from 'uuid-mongodb';
 import { NotifyService } from '@/notify/notify.service';
 import { ConfigModule } from '@/common/config/config.module';
 import { ScenarioDbService } from '@/scenario/scenario-db/scenario-db.service';
 import { EventEmitter2 } from '@nestjs/event-emitter';
 
+function normalizeBsonUUID(process: { _id: MUUID; scenario: MUUID; [_: string]: any }) {
+  const { _id: id, scenario, ...rest } = process;
+
+  return {
+    _id: id.toString(),
+    scenario: scenario.toString(),
+    ...rest,
+  };
+}
+
 describe('ProcessService', () => {
   let module: TestingModule;
   let service: ProcessService;
-  let collection: Collection;
-  let scenarios: ScenarioService;
+  let collection: jest.Mocked<Collection>;
+  let scenarios: jest.Mocked<ScenarioService>;
+  let eventEmitter: jest.Mocked<EventEmitter2>;
 
   beforeEach(async () => {
     module = await Test.createTestingModule({
@@ -31,8 +42,9 @@ describe('ProcessService', () => {
       ],
     }).compile();
 
-    service = module.get<ProcessService>(ProcessService);
-    scenarios = module.get<ScenarioService>(ScenarioService);
+    service = module.get(ProcessService);
+    scenarios = module.get(ScenarioService);
+    eventEmitter = module.get(EventEmitter2);
 
     collection = {
       findOne: jest.fn(),
@@ -173,13 +185,11 @@ describe('ProcessService', () => {
     const process = step(instantiate(scenario), 'start');
 
     it('should save a process', async () => {
-      const replaceOne = jest.spyOn(collection, 'replaceOne').mockResolvedValue({} as any);
-
       await service.save(process);
 
-      expect(replaceOne).toHaveBeenCalled();
-      expect(replaceOne.mock.calls[0][0]._id.toString()).toEqual(process.id);
-      const stored = replaceOne.mock.calls[0][1];
+      expect(collection.replaceOne).toHaveBeenCalled();
+      expect(collection.replaceOne.mock.calls[0][0]._id.toString()).toEqual(process.id);
+      const stored = collection.replaceOne.mock.calls[0][1];
 
       expect(stored._id.toString()).toEqual(process.id);
       expect(stored.scenario.toString()).toEqual(process.scenario.id);
@@ -209,8 +219,6 @@ describe('ProcessService', () => {
 
     it('should instantiate a process', async () => {
       jest.spyOn(scenarios, 'get').mockResolvedValue({ id: scenarioId, ...scenario, _disabled: false });
-      const replaceOne = jest.spyOn(collection, 'replaceOne').mockResolvedValue({} as any);
-
       const process = await service.instantiate(scenario);
 
       expect(process.id).toMatch(
@@ -251,24 +259,23 @@ describe('ProcessService', () => {
   describe('has', () => {
     it('should return true is the process exists', async () => {
       const processId = 'b2d39a1f-88bb-450e-95c5-feeffe95abe6';
-      const countDocuments = jest.spyOn(collection, 'countDocuments').mockResolvedValue(1);
+      collection.countDocuments.mockResolvedValue(1);
 
       const exists = await service.has(processId);
       expect(exists).toBe(true);
 
-      expect(countDocuments).toHaveBeenCalled();
-      expect(countDocuments.mock.calls[0][0]._id.toString()).toEqual(processId);
+      expect(collection.countDocuments).toHaveBeenCalled();
+      expect(collection.countDocuments.mock.calls[0][0]._id.toString()).toEqual(processId);
     });
 
     it("should return false is the process doesn't exist", async () => {
       const processId = 'b2d39a1f-88bb-450e-95c5-feeffe95abe6';
-      const countDocuments = jest.spyOn(collection, 'countDocuments').mockResolvedValue(0);
 
       const exists = await service.has(processId);
       expect(exists).toBe(false);
 
-      expect(countDocuments).toHaveBeenCalled();
-      expect(countDocuments.mock.calls[0][0]._id.toString()).toEqual(processId);
+      expect(collection.countDocuments).toHaveBeenCalled();
+      expect(collection.countDocuments.mock.calls[0][0]._id.toString()).toEqual(processId);
     });
   });
 
@@ -334,7 +341,7 @@ describe('ProcessService', () => {
       const processDocument = { _id: bsonUUID(processId), scenario: bsonUUID(scenarioId), ...processData };
 
       jest.spyOn(scenarios, 'get').mockResolvedValue({ ...scenario, _disabled: false });
-      const findOne = jest.spyOn(collection, 'findOne').mockResolvedValue(processDocument);
+      collection.findOne.mockResolvedValue(processDocument);
 
       const process = await service.get(processId);
 
@@ -365,17 +372,73 @@ describe('ProcessService', () => {
 
       expect(scenarios.get).toHaveBeenCalledWith(scenarioId);
       expect(collection.findOne).toHaveBeenCalled();
-      expect(findOne.mock.calls[0][0]._id.toString()).toEqual(processId);
+      expect(collection.findOne.mock.calls[0][0]._id.toString()).toEqual(processId);
     });
 
     it("should throw an error if the process doesn't exist", async () => {
       const processId = 'b2d39a1f-88bb-450e-95c5-feeffe95abe6';
-      const findOne = jest.spyOn(collection, 'findOne').mockResolvedValue(null);
 
       await expect(service.get(processId)).rejects.toThrow('Process not found');
 
-      expect(findOne).toHaveBeenCalled();
-      expect(findOne.mock.calls[0][0]._id.toString()).toEqual(processId);
+      expect(collection.findOne).toHaveBeenCalled();
+      expect(collection.findOne.mock.calls[0][0]._id.toString()).toEqual(processId);
+    });
+  });
+
+  describe('save', () => {
+    const scenario = normalize({
+      title: 'simple workflow',
+      states: {
+        initial: {
+          on: 'complete',
+          goto: '(done)',
+        },
+      },
+    });
+    const scenarioId = uuid(scenario);
+
+    const process = instantiate(scenario);
+    const { id: processId, scenario: _, actors, ...rest } = process;
+
+    const doc = {
+      _id: bsonUUID(processId),
+      scenario: bsonUUID(scenarioId),
+      actors: Object.entries(actors).map(([key, actor]) => ({ _key: key, ...actor })),
+      ...rest,
+    };
+
+    it('should save a process', async () => {
+      await service.save(process);
+
+      expect(collection.replaceOne).toHaveBeenCalled();
+      expect(collection.replaceOne.mock.calls[0][0]._id.toString()).toEqual(processId);
+      expect(normalizeBsonUUID(collection.replaceOne.mock.calls[0][1] as any)).toEqual(normalizeBsonUUID(doc));
+    });
+  });
+
+  describe('step', () => {
+    const scenario = normalize({
+      title: 'simple workflow',
+      states: {
+        initial: {
+          on: 'complete',
+          goto: '(done)',
+        },
+        done: {},
+      },
+    });
+
+    it('should step through a process', async () => {
+      const save = jest.spyOn(service, 'save');
+      const process = instantiate(scenario);
+
+      const updatedProcess = await service.step(process, 'complete', { key: 'actor' });
+
+      expect(updatedProcess.id).toEqual(process.id);
+      expect(updatedProcess.current.key).toEqual('(done)');
+
+      expect(save).toHaveBeenCalledWith(updatedProcess);
+      expect(eventEmitter.emit).toBeCalledWith('process.stepped', updatedProcess);
     });
   });
 });