summaryrefslogtreecommitdiff
path: root/worker/executor/pipeline.ts
blob: a1aa7c32c9b07d2863025cedb82288994399bc13 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import {
  Either,
  type IEither,
  type ITraceable,
  type LogMetricTraceSupplier,
  Metric,
  TraceUtil,
} from "@emprespresso/pengueno";
import type { Job, JobArgT, Pipeline } from "@emprespresso/ci-model";
import { executeJob } from "./mod.ts";

const pipelinesMetric = Metric.fromName("pipelines");
export const executePipeline = (
  tPipeline: ITraceable<Pipeline, LogMetricTraceSupplier>,
  baseEnv?: JobArgT,
): Promise<IEither<Error, void>> =>
  tPipeline.bimap(TraceUtil.withFunctionTrace(executePipeline))
    .bimap(TraceUtil.withMetricTrace(pipelinesMetric))
    .map(async (tJobs): Promise<IEither<Error, void>> => {
      for (const [i, serialStage] of tJobs.get().serialJobs.entries()) {
        tJobs.trace.trace(
          `executing stage ${i}. do your best little stage :>\n${serialStage}`,
        );
        const jobResults = await Promise.all(
          serialStage.parallelJobs.map((job) =>
            tJobs.bimap((_) => [job, `stage ${i}`])
              .map((tJob) =>
                <Job> ({
                  ...tJob.get(),
                  arguments: { ...baseEnv, ...tJob.get().arguments },
                })
              )
              .map(executeJob)
              .peek(
                TraceUtil.promiseify((tEitherJobOutput) =>
                  tEitherJobOutput.get().mapRight((stdout) =>
                    tEitherJobOutput.trace.addTrace("STDOUT").trace(stdout)
                  )
                ),
              )
              .get()
          ),
        );
        const failures = jobResults.filter((e) => e.fold((err) => !!err));
        if (failures.length > 0) {
          tJobs.trace.trace(pipelinesMetric.failure);
          return Either.left(new Error(failures.toString()));
        }
      }
      tJobs.trace.trace(pipelinesMetric.success);
      return Either.right(undefined);
    })
    .get();