Move functionality from benchmarks to shared location

Author: bigfluffycookie

packages/server/api/src/app/benchmark/benchmark-status.service.ts

@@ -6,10 +6,13 @@ import {
   BenchmarkStatusResponse,
   BenchmarkWorkflowStatusItem,
   ErrorCode,
-  FlowRunStatus,
 } from '@openops/shared';
 import { IsNull } from 'typeorm';
-import { flowRunRepo } from '../flows/flow-run/flow-run-service';
+import {
+  FlowRunSummary,
+  getLatestRunByFlowId,
+  resolveRunStatus,
+} from '../flows/flow-run/flow-run-status.service';
 import { benchmarkFlowRepo } from './benchmark-flow.repo';
 import { benchmarkRepo } from './benchmark.repo';
 
@@ -22,30 +25,6 @@ type FlowRow = {
 
 type FlowRowWithBenchmarkId = FlowRow & { benchmarkId: string };
 
-type FlowRunSummary = {
-  id: string;
-  status: FlowRunStatus;
-  finishTime?: string;
-};
-
-function mapFlowRunStatusToBenchmarkStatus(
-  status: FlowRunStatus,
-): BenchmarkStatus {
-  switch (status) {
-    case FlowRunStatus.RUNNING:
-    case FlowRunStatus.PAUSED:
-    case FlowRunStatus.SCHEDULED:
-      return BenchmarkStatus.RUNNING;
-    case FlowRunStatus.FAILED:
-    case FlowRunStatus.INTERNAL_ERROR:
-    case FlowRunStatus.TIMEOUT:
-    case FlowRunStatus.IGNORED:
-      return BenchmarkStatus.FAILED;
-    default:
-      return BenchmarkStatus.SUCCEEDED;
-  }
-}
-
 async function fetchFlowRowsByBenchmarkIds(
   benchmarkIds: string[],
 ): Promise<FlowRowWithBenchmarkId[]> {
@@ -80,63 +59,17 @@ function buildWorkflowStatusItems(
       isOrchestrator: row.isOrchestrator,
       isCleanup: row.isCleanup,
       runStatus: latestRun
-        ? mapFlowRunStatusToBenchmarkStatus(latestRun.status)
+        ? resolveRunStatus(latestRun.status)
         : BenchmarkStatus.CREATED,
       runId: latestRun?.id,
     };
   });
 }
 
-async function getLatestRunByFlowId(
-  flowIds: string[],
-  projectId: string,
-): Promise<Record<string, FlowRunSummary | undefined>> {
-  const rows = await flowRunRepo()
-    .createQueryBuilder('fr')
-    .distinctOn(['fr.flowId'])
-    .select('fr.flowId', 'flowId')
-    .addSelect('fr.id', 'id')
-    .addSelect('fr.status', 'status')
-    .addSelect('fr.finishTime', 'finishTime')
-    .where('fr.projectId = :projectId', { projectId })
-    .andWhere('fr.flowId IN (:...flowIds)', { flowIds })
-    .orderBy('fr.flowId', 'ASC')
-    .addOrderBy('fr.created', 'DESC')
-    .addOrderBy('fr.id', 'DESC')
-    .getRawMany<{
-      id: string;
-      flowId: string;
-      status: FlowRunStatus;
-      finishTime?: string;
-    }>();
-
-  return mapLatestRuns(flowIds, rows);
-}
-
-function mapLatestRuns(
-  flowIds: string[],
-  rows: {
-    flowId: string;
-    id: string;
-    status: FlowRunStatus;
-    finishTime?: string;
-  }[],
-): Record<string, FlowRunSummary | undefined> {
-  const result: Record<string, FlowRunSummary | undefined> = Object.fromEntries(
-    flowIds.map((fid) => [fid, undefined]),
-  );
-  for (const r of rows) {
-    result[r.flowId] = { id: r.id, status: r.status, finishTime: r.finishTime };
-  }
-  return result;
-}
-
 function resolveOrchestratorStatus(
   run: FlowRunSummary | undefined,
 ): BenchmarkStatus {
-  return run
-    ? mapFlowRunStatusToBenchmarkStatus(run.status)
-    : BenchmarkStatus.CREATED;
+  return run ? resolveRunStatus(run.status) : BenchmarkStatus.CREATED;
 }
 
 async function resolveStatusByBenchmarkId(

packages/server/api/src/app/benchmark/create-benchmark.service.ts

@@ -8,11 +8,11 @@ import {
   ErrorCode,
   Folder,
 } from '@openops/shared';
-import fs from 'node:fs/promises';
 import { IsNull } from 'typeorm';
 import { getConnectionsWithBlockSupport } from '../app-connection/connections-with-block-support';
 import {
   bulkCreateAndPublishFlows,
+  loadWorkflowTemplate,
   type WorkflowTemplate,
 } from '../flows/flow/flow-bulk-create';
 import { flowService } from '../flows/flow/flow.service';
@@ -137,16 +137,6 @@ export async function deleteFlowsForExistingBenchmark(params: {
   );
 }
 
-async function loadWorkflowTemplate(
-  filePath: string,
-): Promise<WorkflowTemplate> {
-  const content = await fs.readFile(filePath, 'utf-8');
-  const parsed = JSON.parse(content) as {
-    template: WorkflowTemplate['template'];
-  };
-  return { template: parsed.template };
-}
-
 type CategorizedWorkflowTemplates = {
   orchestrator: WorkflowTemplate;
   cleanup: WorkflowTemplate;

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

@@ -0,0 +1,62 @@
+import { FlowRunStatus, SimplifiedRunStatus } from '@openops/shared';
+import { flowRunRepo } from './flow-run-service';
+
+export type FlowRunSummary = {
+  id: string;
+  status: FlowRunStatus;
+  finishTime?: string;
+};
+
+export function resolveRunStatus(
+  status: FlowRunStatus,
+): SimplifiedRunStatus {
+  switch (status) {
+    case FlowRunStatus.RUNNING:
+    case FlowRunStatus.PAUSED:
+    case FlowRunStatus.SCHEDULED:
+      return SimplifiedRunStatus.RUNNING;
+    case FlowRunStatus.FAILED:
+    case FlowRunStatus.INTERNAL_ERROR:
+    case FlowRunStatus.TIMEOUT:
+    case FlowRunStatus.IGNORED:
+      return SimplifiedRunStatus.FAILED;
+    default:
+      return SimplifiedRunStatus.SUCCEEDED;
+  }
+}
+
+export async function getLatestRunByFlowId(
+  flowIds: string[],
+  projectId: string,
+): Promise<Record<string, FlowRunSummary | undefined>> {
+  if (flowIds.length === 0) {
+    return {};
+  }
+
+  const rows = await flowRunRepo()
+    .createQueryBuilder('fr')
+    .distinctOn(['fr.flowId'])
+    .select('fr.flowId', 'flowId')
+    .addSelect('fr.id', 'id')
+    .addSelect('fr.status', 'status')
+    .addSelect('fr.finishTime', 'finishTime')
+    .where('fr.projectId = :projectId', { projectId })
+    .andWhere('fr.flowId IN (:...flowIds)', { flowIds })
+    .orderBy('fr.flowId', 'ASC')
+    .addOrderBy('fr.created', 'DESC')
+    .addOrderBy('fr.id', 'DESC')
+    .getRawMany<{
+      id: string;
+      flowId: string;
+      status: FlowRunStatus;
+      finishTime?: string;
+    }>();
+
+  const result: Record<string, FlowRunSummary | undefined> = Object.fromEntries(
+    flowIds.map((fid) => [fid, undefined]),
+  );
+  for (const r of rows) {
+    result[r.flowId] = { id: r.id, status: r.status, finishTime: r.finishTime };
+  }
+  return result;
+}

packages/server/api/src/app/flows/flow/flow-bulk-create.ts

@@ -7,9 +7,20 @@ import {
   flowHelper,
   openOpsId,
 } from '@openops/shared';
+import fs from 'node:fs/promises';
 import { EntityManager } from 'typeorm';
 import { flowRepo } from './flow.repo';
 
+export async function loadWorkflowTemplate(
+  filePath: string,
+): Promise<WorkflowTemplate> {
+  const content = await fs.readFile(filePath, 'utf-8');
+  const parsed = JSON.parse(content) as {
+    template: WorkflowTemplate['template'];
+  };
+  return { template: parsed.template };
+}
+
 export type WorkflowTemplate = {
   template: {
     displayName: string;

packages/shared/src/lib/benchmark/dto/benchmark-status-response.ts

@@ -1,17 +1,13 @@
 import { Static, Type } from '@sinclair/typebox';
+import { SimplifiedRunStatus } from '../../common/run-status';
 import { BenchmarkWorkflowBase } from './create-benchmark-response';
 
-export enum BenchmarkStatus {
-  CREATED = 'CREATED',
-  RUNNING = 'RUNNING',
-  SUCCEEDED = 'SUCCEEDED',
-  FAILED = 'FAILED',
-}
+export { SimplifiedRunStatus as BenchmarkStatus };
 
 export const BenchmarkWorkflowStatusItem = Type.Intersect([
   BenchmarkWorkflowBase,
   Type.Object({
-    runStatus: Type.Enum(BenchmarkStatus),
+    runStatus: Type.Enum(SimplifiedRunStatus),
     runId: Type.Optional(Type.String()),
   }),
 ]);
@@ -22,7 +18,7 @@ export type BenchmarkWorkflowStatusItem = Static<
 
 export const BenchmarkStatusResponse = Type.Object({
   benchmarkId: Type.String(),
-  status: Type.Enum(BenchmarkStatus),
+  status: Type.Enum(SimplifiedRunStatus),
   workflows: Type.Array(BenchmarkWorkflowStatusItem),
   lastRunId: Type.Optional(Type.String()),
   lastRunFinishedAt: Type.Optional(Type.String()),

packages/shared/src/lib/common/index.ts

@@ -1,4 +1,5 @@
 export * from './base-model';
 export * from './locale';
+export * from './run-status';
 export * from './security';
 export * from './utils';

packages/shared/src/lib/common/run-status.ts

@@ -0,0 +1,6 @@
+export enum SimplifiedRunStatus {
+  CREATED = 'CREATED',
+  RUNNING = 'RUNNING',
+  SUCCEEDED = 'SUCCEEDED',
+  FAILED = 'FAILED',
+}