fix(worker): compile actor variables in sent notifications and exclude actor from fan-out fixes NV-8143

apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.usecase.spec.ts

@@ -114,4 +114,48 @@ describe('ExecuteBridgeJob - redundant workflow lookup', () => {
     expect(executeBridgeRequest.execute.calledOnce).to.equal(true);
     expect(result).to.deep.equal({ outputs: {}, options: {} });
   });
+
+  it('passes actor to the bridge event when provided in variables', async () => {
+    const { usecase, executeBridgeRequest } = buildUsecase();
+    const actor = {
+      subscriberId: 'actor-subscriber-01',
+      firstName: 'Actor',
+      lastName: 'User',
+      email: 'actor@example.com',
+    };
+
+    const command = {
+      environmentId: 'env_1',
+      organizationId: 'org_1',
+      userId: 'user_1',
+      identifier: 'wf-identifier',
+      jobId: 'job_1',
+      job: {
+        _id: 'job_1',
+        _templateId: 'tpl_1',
+        _parentId: undefined,
+        _environmentId: 'env_1',
+        _organizationId: 'org_1',
+        step: { stepId: 'step_1', uuid: 'step_1' },
+      },
+      variables: {
+        subscriber: { subscriberId: 'recipient-01' },
+        actor,
+        payload: {},
+        env: { name: 'Development', type: 'dev' },
+      },
+      workflow: {
+        _id: 'tpl_1',
+        type: ResourceTypeEnum.BRIDGE,
+        origin: ResourceOriginEnum.NOVU_CLOUD,
+        triggers: [{ identifier: 'wf-identifier' }],
+      },
+    } as never;
+
+    await usecase.execute(command);
+
+    expect(executeBridgeRequest.execute.calledOnce).to.equal(true);
+    const bridgeRequest = executeBridgeRequest.execute.firstCall.args[0];
+    expect(bridgeRequest.event.actor).to.deep.equal(actor);
+  });
 });

apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.usecase.ts

@@ -110,7 +110,7 @@ export class ExecuteBridgeJob {
       throw new Error(`Bridge URL is not set for environment id: ${environment._id}`);
     }
 
-    const { subscriber, payload: originalPayload, context, env } = command.variables || {};
+    const { subscriber, actor, payload: originalPayload, context, env } = command.variables || {};
     const payload = this.normalizePayload(originalPayload);
     const state = await this.generateState(command);
 
@@ -124,6 +124,7 @@ export class ExecuteBridgeJob {
       controls: variablesStores ?? {},
       state,
       subscriber: subscriber ?? {},
+      ...(actor && { actor }),
       context: context ?? {},
       // biome-ignore lint/style/noNonNullAssertion: <explanation> we always have env.type and env.name
       env: env!,

libs/application-generic/src/usecases/trigger-broadcast/trigger-broadcast.usecase.ts

@@ -43,6 +43,10 @@ export class TriggerBroadcast extends TriggerBase {
         {},
         subscriberFetchBatchSize
       )) {
+        if (command.actor?.subscriberId && subscriber.subscriberId === command.actor.subscriberId) {
+          continue;
+        }
+
         subscribers.push(subscriber);
         if (subscribers.length === subscriberFetchBatchSize) {
           await this.sendToProcessSubscriberService(command, subscribers, SubscriberSourceEnum.BROADCAST);