Fix notify service for core lib v1.2

Author: jasny

src/notify/amqp/ampq.service.ts

@@ -1,8 +1,7 @@
 import { Inject, Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
-import { etag, Notify, Process } from '@letsflow/core/process';
+import { createMessage, etag, Notify, Process } from '@letsflow/core/process';
 import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
 import { ConfigService } from '@/common/config/config.service';
-import { createMessage } from '../utils/message';
 import { NotifyProvider } from '../notify-provider.interface';
 
 const DEFAULT_APPID = 'letsflow';
@@ -88,7 +87,7 @@ export class AmqpService implements NotifyProvider, OnModuleDestroy {
     const { url: _, exchange, routingKey, responseTimeout, reply, ...options } = settings;
 
     const channel = await this.getChannel(args.service);
-    const message = args.message ?? createMessage(process, args.trigger);
+    const message = args.message ?? createMessage(process, args.service);
 
     options.appId ??= DEFAULT_APPID;
     options.messageId = etag(process);

src/notify/amqp/amqp.service.spec.ts

@@ -2,7 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { AmqpService } from './ampq.service';
 import { ConfigService } from '@/common/config/config.service';
 import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
-import { Notify, Process } from '@letsflow/core/process';
+import { Process } from '@letsflow/core/process';
 
 describe('AmqpService', () => {
   let service: AmqpService;
@@ -14,7 +14,8 @@ describe('AmqpService', () => {
   const process = {
     id: '00000000-0000-0000-0001-000000000001',
     current: {
-      actions: [{ key: 'next' }],
+      actions: [{ key: 'next', actor: ['service:pushService', 'service:replyService'] }],
+      instructions: { 'service:pushService': 'Go to next' } as Record<string, string>,
     },
     events: [{ hash: '1234' }],
   } as Process;
@@ -119,7 +120,7 @@ describe('AmqpService', () => {
         },
       });
 
-      const args = { service: 'pushService', trigger: 'next' } as Notify;
+      const args = { service: 'pushService', after: 0 };
 
       await service.notify(process, args);
 
@@ -128,7 +129,8 @@ describe('AmqpService', () => {
         'test-key',
         JSON.stringify({
           process: '00000000-0000-0000-0001-000000000001',
-          action: { key: 'next' },
+          actions: [{ key: 'next', actor: ['service:pushService', 'service:replyService'] }],
+          instructions: 'Go to next',
           etag: '1234',
         }),
         {
@@ -161,7 +163,7 @@ describe('AmqpService', () => {
         });
       });
 
-      const args = { service: 'replyService', trigger: 'next' } as Notify;
+      const args = { service: 'replyService', after: 0 };
 
       const result = await service.notify(process, args);
 
@@ -172,7 +174,7 @@ describe('AmqpService', () => {
         '',
         JSON.stringify({
           process: '00000000-0000-0000-0001-000000000001',
-          action: { key: 'next' },
+          actions: [{ key: 'next', actor: ['service:pushService', 'service:replyService'] }],
           etag: '1234',
         }),
         {
@@ -188,7 +190,7 @@ describe('AmqpService', () => {
     });
 
     it('should throw an error if the response timeout is exceeded', async () => {
-      const args = { service: 'replyService', trigger: 'next' } as Notify;
+      const args = { service: 'replyService', after: 0 };
 
       await expect(service.notify(process, args)).rejects.toThrow('Response timeout exceeded');
     });

src/notify/utils/message.ts

@@ -11,7 +11,8 @@ describe('WebhookService', () => {
   const process = {
     id: '00000000-0000-0000-0001-000000000001',
     current: {
-      actions: [{ key: 'next' }],
+      actions: [{ key: 'next', actor: ['service:test'] }],
+      instructions: { 'service:test': 'Go to next' } as Record<string, string>,
     },
     events: [{ hash: '1234' }],
   } as Process;
@@ -26,7 +27,7 @@ describe('WebhookService', () => {
           provide: ConfigService,
           useValue: {
             get: jest.fn(() => ({
-              testService: {
+              test: {
                 url: 'https://example.com/webhook',
                 timeout: 10000,
               },
@@ -59,18 +60,18 @@ describe('WebhookService', () => {
 
     it('should throw an error if the url is missing in settings', async () => {
       jest.spyOn(configService, 'get').mockReturnValueOnce({
-        testService: {},
+        test: {},
       });
 
-      await expect(service.notify(process, { service: 'testService' } as Notify)).rejects.toThrow(
-        "Service 'testService' is missing url setting",
+      await expect(service.notify(process, { service: 'test' } as Notify)).rejects.toThrow(
+        "Service 'test' is missing url setting",
       );
     });
 
     it('should send a message with correct headers and body', async () => {
       fetchMock.mockResolvedValue({ ok: true, status: 202 });
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
       const result = await service.notify(process, args);
 
       expect(fetchMock).toHaveBeenCalledWith('https://example.com/webhook', {
@@ -79,7 +80,8 @@ describe('WebhookService', () => {
         headers: { 'Content-Type': 'application/json' },
         body: JSON.stringify({
           process: '00000000-0000-0000-0001-000000000001',
-          action: { key: 'next' },
+          actions: [{ key: 'next', actor: ['service:test'] }],
+          instructions: 'Go to next',
           etag: '1234',
         }),
         signal: expect.any(AbortSignal),
@@ -96,7 +98,7 @@ describe('WebhookService', () => {
         text: jest.fn().mockResolvedValue('Server error'),
       });
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
 
       await expect(service.notify(process, args)).rejects.toThrow(
         'Webhook failed with status 500 Internal Server Error: Server error',
@@ -111,7 +113,7 @@ describe('WebhookService', () => {
         text: jest.fn().mockResolvedValue('Success'),
       });
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
       const result = await service.notify(process, args);
 
       expect(result).toBe('Success');
@@ -125,7 +127,7 @@ describe('WebhookService', () => {
         json: jest.fn().mockResolvedValue({ success: true }),
       });
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
       const result = await service.notify(process, args);
 
       expect(result).toEqual({ success: true });

src/notify/webhook/webhook.service.spec.ts

@@ -1,7 +1,6 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { ConfigService } from '@/common/config/config.service';
-import { Notify, Process } from '@letsflow/core/process';
-import { createMessage } from '@/notify/utils/message';
+import { createMessage, Notify, Process } from '@letsflow/core/process';
 import { NotifyProvider } from '@/notify/notify-provider.interface';
 
 interface WebhookSettings extends RequestInit {
@@ -28,7 +27,7 @@ export class WebhookService implements NotifyProvider {
       throw new Error(`Service '${args.service}' is missing url setting`);
     }
 
-    const message = args.message ?? createMessage(process, args.trigger);
+    const message = args.message ?? createMessage(process, args.service);
     settings.headers ??= {};
     settings.headers['Content-Type'] ??= typeof message === 'string' ? 'text/plain' : 'application/json';
 

src/notify/webhook/webhook.service.ts

@@ -2,7 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
 import { ZeromqService } from './zeromq.service';
 import { ConfigService } from '@/common/config/config.service';
 import { Push } from 'zeromq';
-import { Notify, Process } from '@letsflow/core/process';
+import { Process } from '@letsflow/core/process';
 
 describe('ZeromqService', () => {
   let service: ZeromqService;
@@ -12,7 +12,8 @@ describe('ZeromqService', () => {
   const process = {
     id: '00000000-0000-0000-0001-000000000001',
     current: {
-      actions: [{ key: 'next' }],
+      actions: [{ key: 'next', actor: ['service:test'] }],
+      instructions: { 'service:test': 'Go to next' } as Record<string, string>,
     },
     events: [{ hash: '1234' }],
   } as Process;
@@ -27,7 +28,7 @@ describe('ZeromqService', () => {
           provide: ConfigService,
           useValue: {
             get: jest.fn(() => ({
-              testService: { type: 'push' },
+              test: { type: 'push' },
             })),
           },
         },
@@ -49,7 +50,7 @@ describe('ZeromqService', () => {
   describe('onModuleDestroy', () => {
     it('should close all sockets', () => {
       const mockSocket = { close: jest.fn() };
-      service['sockets'].set('testService', mockSocket as any);
+      service['sockets'].set('test', mockSocket as any);
 
       service.onModuleDestroy();
 
@@ -60,9 +61,9 @@ describe('ZeromqService', () => {
   describe('getSocket', () => {
     it('should return an existing socket if it exists', () => {
       const mockSocket = {};
-      service['sockets'].set('testService', mockSocket as any);
+      service['sockets'].set('test', mockSocket as any);
 
-      const result = service.getSocket('testService');
+      const result = service.getSocket('test');
 
       expect(result).toBe(mockSocket);
     });
@@ -71,7 +72,7 @@ describe('ZeromqService', () => {
       const mockPushSocket = new Push();
       createSocket.mockReturnValue(mockPushSocket);
 
-      const result = service.getSocket('testService');
+      const result = service.getSocket('test');
 
       expect(createSocket).toHaveBeenCalledWith({ type: 'push' });
       expect(result).toBe(mockPushSocket);
@@ -89,14 +90,15 @@ describe('ZeromqService', () => {
       const mockPushSocket = { send: jest.fn() };
       createSocket.mockReturnValue(mockPushSocket);
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
 
       const result = await service.notify(process, args);
 
       expect(mockPushSocket.send).toHaveBeenCalledWith(
         JSON.stringify({
           process: '00000000-0000-0000-0001-000000000001',
-          action: { key: 'next' },
+          actions: [{ key: 'next', actor: ['service:test'] }],
+          instructions: 'Go to next',
           etag: '1234',
         }),
       );
@@ -108,7 +110,7 @@ describe('ZeromqService', () => {
       const mockPushSocket = { send: jest.fn().mockRejectedValue({ message: 'Test error' }) };
       createSocket.mockReturnValue(mockPushSocket);
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
 
       expect(service.notify(process, args)).rejects.toEqual({ message: 'Test error' });
     });
@@ -120,14 +122,15 @@ describe('ZeromqService', () => {
       };
       createSocket.mockReturnValue(mockReplySocket);
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
 
       const result = await service.notify(process, args);
 
       expect(mockReplySocket.send).toHaveBeenCalledWith(
         JSON.stringify({
           process: '00000000-0000-0000-0001-000000000001',
-          action: { key: 'next' },
+          actions: [{ key: 'next', actor: ['service:test'] }],
+          instructions: 'Go to next',
           etag: '1234',
         }),
       );
@@ -144,7 +147,7 @@ describe('ZeromqService', () => {
 
       createSocket.mockReturnValue(mockReplySocket);
 
-      const args = { service: 'testService', trigger: 'next' } as Notify;
+      const args = { service: 'test', after: 0 };
 
       expect(service.notify(process, args)).rejects.toEqual({ message: 'Test error' });
     });

src/notify/zeromq/zeromq.service.spec.ts

@@ -1,8 +1,7 @@
 import { Inject, Injectable, OnModuleDestroy } from '@nestjs/common';
-import { Notify, Process } from '@letsflow/core/process';
+import { createMessage, Notify, Process } from '@letsflow/core/process';
 import { Push, Reply, SocketOptions } from 'zeromq';
 import { ConfigService } from '@/common/config/config.service';
-import { createMessage } from '../utils/message';
 import { NotifyProvider } from '../notify-provider.interface';
 
 export type ZeromqOptions =
@@ -42,12 +41,12 @@ export class ZeromqService implements NotifyProvider, OnModuleDestroy {
 
   async notify(process: Process, args: Notify): Promise<any> {
     const socket = this.getSocket(args.service);
-    const message = args.message ?? createMessage(process, args.trigger);
+    const message = args.message ?? createMessage(process, args.service);
 
     await socket.send(typeof message === 'string' ? message : JSON.stringify(message));
 
+    // Use `in` and not `instanceof` for compatibility with Jest mocks
     if ('receive' in socket) {
-      // Don't use instanceof for compatibility with Jest mocks
       const [response] = await socket.receive();
       return response;
     }

src/notify/zeromq/zeromq.service.ts

@@ -236,7 +236,6 @@ describe('ProcessService', () => {
           description: '',
           title: 'complete',
           key: 'complete',
-          if: true,
           response: {},
         },
       ]);
@@ -424,7 +423,7 @@ describe('ProcessService', () => {
     });
 
     it('should step through a process', async () => {
-      const save = jest.spyOn(service, 'save');
+      const save = jest.spyOn(service, 'save').mockResolvedValue(undefined);
       const process = instantiate(scenario);
 
       const updatedProcess = await service.step(process, 'complete', { key: 'actor' });

src/process/process.service.spec.ts

@@ -211,7 +211,10 @@ export class ProcessService {
     const updatedProcess = step(process, action, actor, response);
     await this.save(updatedProcess);
 
-    if (process.current.timestamp.getTime() !== updatedProcess.current.timestamp.getTime()) {
+    if (
+      process.current.key !== updatedProcess.current.key ||
+      process.current.timestamp.getTime() !== updatedProcess.current.timestamp.getTime()
+    ) {
       this.eventEmitter.emit('process.stepped', updatedProcess);
     }