Add new column for trigger source in flow_run table (#1096)

Author: bigfluffycookie

packages/server/api/src/app/database/migrations/1754489349007-AddTriggerSourceToFlowRun.ts

@@ -0,0 +1,29 @@
+import { FlowRunTriggerSource } from '@openops/shared';
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddTriggerSourceToFlowRun1754489349007
+  implements MigrationInterface
+{
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE flow_run 
+      ADD COLUMN IF NOT EXISTS "triggerSource" varchar;
+
+      UPDATE flow_run 
+      SET "triggerSource" = CASE
+        WHEN environment = 'TESTING' THEN '${FlowRunTriggerSource.TEST_RUN}'
+        ELSE '${FlowRunTriggerSource.TRIGGERED}'
+      END
+      WHERE "triggerSource" IS NULL;
+
+      ALTER TABLE flow_run 
+      ALTER COLUMN "triggerSource" SET NOT NULL;
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE flow_run DROP COLUMN IF EXISTS "triggerSource";
+    `);
+  }
+}

packages/server/api/src/app/database/postgres-connection.ts

@@ -29,6 +29,7 @@ import { AddInputToTestOutputTable1750933522145 } from './migrations/17509335221
 import { AddInputToTriggerEventTable1751282188913 } from './migrations/1751282188913-AddInputToTriggerEventTable';
 import { AddSuccessToFlowStepTestOutputTable1752487641303 } from './migrations/1752487641303-AddSuccessToFlowStepTestOutputTable';
 import { UpdateUmbrellaRecommendationTypesAcrossFlowVersions1752758891771 } from './migrations/1752758891771-UpdateUmbrellaRecommendationTypesAcrossFlowVersions';
+import { AddTriggerSourceToFlowRun1754489349007 } from './migrations/1754489349007-AddTriggerSourceToFlowRun';
 
 const getSslConfig = (): boolean | TlsOptions => {
   const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL);
@@ -66,6 +67,7 @@ const getMigrations = (): (new () => MigrationInterface)[] => {
     AddInputToTriggerEventTable1751282188913,
     AddSuccessToFlowStepTestOutputTable1752487641303,
     UpdateUmbrellaRecommendationTypesAcrossFlowVersions1752758891771,
+    AddTriggerSourceToFlowRun1754489349007,
   ];
 };
 

packages/server/api/src/app/flows/flow-run/flow-run-entity.ts

@@ -1,4 +1,10 @@
-import { File, Flow, FlowRun, Project } from '@openops/shared';
+import {
+  File,
+  Flow,
+  FlowRun,
+  FlowRunTriggerSource,
+  Project,
+} from '@openops/shared';
 import { EntitySchema } from 'typeorm';
 import {
   ARRAY_COLUMN_TYPE,
@@ -36,6 +42,10 @@ export const FlowRunEntity = new EntitySchema<FlowRunSchema>({
     status: {
       type: String,
     },
+    triggerSource: {
+      type: String,
+      enum: Object.values(FlowRunTriggerSource),
+    },
     terminationReason: {
       type: String,
       nullable: true,

packages/server/api/src/app/flows/flow-run/flow-run-service.ts

@@ -13,6 +13,7 @@ import {
   FlowRun,
   FlowRunId,
   FlowRunStatus,
+  FlowRunTriggerSource,
   FlowVersionId,
   isEmpty,
   isNil,
@@ -47,8 +48,15 @@ export const flowRunRepo = repoFactory<FlowRun>(FlowRunEntity);
 const getFlowRunOrCreate = async (
   params: GetOrCreateParams,
 ): Promise<Partial<FlowRun>> => {
-  const { id, projectId, flowId, flowVersionId, flowDisplayName, environment } =
-    params;
+  const {
+    id,
+    projectId,
+    flowId,
+    flowVersionId,
+    flowDisplayName,
+    environment,
+    triggerSource,
+  } = params;
 
   if (id) {
     return flowRunService.getOneOrThrow({
@@ -65,6 +73,7 @@ const getFlowRunOrCreate = async (
     environment,
     flowDisplayName,
     startTime: new Date().toISOString(),
+    triggerSource,
   };
 };
 
@@ -254,6 +263,7 @@ export const flowRunService = {
       progressUpdateType,
       executionType,
       environment: RunEnvironment.PRODUCTION,
+      triggerSource: flowRunToResume.triggerSource,
     });
   },
   async updateStatus({
@@ -302,6 +312,7 @@ export const flowRunService = {
     synchronousHandlerId,
     progressUpdateType,
     executionCorrelationId,
+    triggerSource,
   }: StartParams): Promise<FlowRun> {
     const flowVersion = await flowVersionService.getOneOrThrow(flowVersionId);
 
@@ -317,6 +328,7 @@ export const flowRunService = {
       flowVersionId: flowVersion.id,
       environment,
       flowDisplayName: flowVersion.displayName,
+      triggerSource,
     });
 
     flowRun.status = FlowRunStatus.SCHEDULED;
@@ -364,6 +376,7 @@ export const flowRunService = {
       synchronousHandlerId: webhookResponseWatcher.getServerId(),
       executionCorrelationId: nanoid(),
       progressUpdateType: ProgressUpdateType.TEST_FLOW,
+      triggerSource: FlowRunTriggerSource.TEST_RUN,
     });
   },
 
@@ -492,6 +505,7 @@ type GetOrCreateParams = {
   flowVersionId: FlowVersionId;
   flowDisplayName: string;
   environment: RunEnvironment;
+  triggerSource: FlowRunTriggerSource;
 };
 
 type ListParams = {
@@ -520,6 +534,7 @@ type StartParams = {
   executionCorrelationId: string;
   progressUpdateType: ProgressUpdateType;
   executionType: ExecutionType;
+  triggerSource: FlowRunTriggerSource;
 };
 
 type TestParams = {

packages/server/api/src/app/workers/worker-controller.ts

@@ -15,6 +15,7 @@ import {
 } from '@openops/server-shared';
 import {
   ExecutionType,
+  FlowRunTriggerSource,
   openOpsId,
   PrincipalType,
   ProgressUpdateType,
@@ -162,6 +163,7 @@ export const flowWorkerController: FastifyPluginAsyncTypebox = async (app) => {
           executionCorrelationId,
           executionType: ExecutionType.BEGIN,
           progressUpdateType,
+          triggerSource: FlowRunTriggerSource.TRIGGERED,
         }),
       );
       return Promise.all(createFlowRuns);
@@ -194,6 +196,7 @@ export const flowWorkerController: FastifyPluginAsyncTypebox = async (app) => {
         executionCorrelationId: data.executionCorrelationId,
         environment: RunEnvironment.PRODUCTION,
         progressUpdateType: data.progressUpdateType ?? ProgressUpdateType.NONE,
+        triggerSource: FlowRunTriggerSource.TRIGGERED,
       });
     },
   );

packages/server/api/test/helpers/mocks/index.ts

@@ -8,6 +8,7 @@ import {
   FlowImportTemplate,
   FlowRun,
   FlowRunStatus,
+  FlowRunTriggerSource,
   FlowStatus,
   FlowVersion,
   FlowVersionState,
@@ -202,6 +203,7 @@ export const createMockFlowRun = (flowRun?: Partial<FlowRun>): FlowRun => {
     finishTime: flowRun?.finishTime ?? faker.date.recent().toISOString(),
     environment:
       flowRun?.environment ?? faker.helpers.enumValue(RunEnvironment),
+    triggerSource: flowRun?.triggerSource ?? FlowRunTriggerSource.TEST_RUN,
   };
 };
 

packages/shared/src/lib/flow-run/flow-run.ts

@@ -20,6 +20,12 @@ export type FlowRetryPayload = {
   strategy: FlowRetryStrategy;
 };
 
+export enum FlowRunTriggerSource {
+  TRIGGERED = 'TRIGGERED',
+  TEST_RUN = 'TEST_RUN',
+  MANUAL_RUN = 'MANUAL_RUN',
+}
+
 export const FlowRun = Type.Object({
   ...BaseModelSchema,
   projectId: Type.String(),
@@ -32,6 +38,7 @@ export const FlowRun = Type.Object({
   logsFileId: Nullable(Type.String()),
   tasks: Type.Optional(Type.Number()),
   status: Type.Enum(FlowRunStatus),
+  triggerSource: Type.Enum(FlowRunTriggerSource),
   duration: Type.Optional(Type.Number()),
   startTime: Type.String(),
   finishTime: Type.Optional(Type.String()),