ApiCore/ApiCore-Backend/src/system/job-run.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { JobRun } from '../../generated/prisma';
import { loggableError } from '../common/loggable-error';
import { JobResult } from '../jobs/job';
import {
  isJobName,
  JOB_FAILURE_PERMISSION,
  JOB_LABELS_DE,
  JOB_NAMES,
  JobName,
} from '../jobs/job-names';
import { NotificationsService } from '../notifications/notifications.service';
import { PrismaService } from '../prisma/prisma.service';

/** JobRun.error of a run whose pod ended before it wrote a result. */
export const JOB_ABORTED_ERROR = 'aborted (pod ended without result)';

/** Notification.relatedType of a job failure; relatedId is the JobRun id. */
const JOB_RUN_RELATED_TYPE = 'JobRun';

const NEWEST_FIRST = [{ startedAt: 'desc' as const }, { id: 'desc' as const }];

type RunOutcome =
  | { status: 'SUCCEEDED'; result: JobResult }
  | { status: 'FAILED'; error: string };

/**
 * Bookkeeping of the CronJob runs: every run is one JobRun row (RUNNING,
 * then SUCCEEDED or FAILED), and a failed run notifies the holders of the
 * job's permission (JOB_FAILURE_PERMISSION).
 */
@Injectable()
export class JobRunService {
  private readonly logger = new Logger(JobRunService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly notifications: NotificationsService,
  ) {}

  /**
   * Runs `work` as one run of `job`. Resolves with the finished JobRun - a
   * failure of `work` is recorded, not rethrown. Throws only if the run
   * itself cannot be recorded.
   */
  async run(job: JobName, work: () => Promise<JobResult>): Promise<JobRun> {
    const started = await this.prisma.jobRun.create({ data: { job } });
    let result: JobResult;
    try {
      result = await work();
    } catch (error: unknown) {
      return this.fail(job, started.id, error);
    }
    this.logger.log(`Job "${job}" succeeded: ${JSON.stringify(result)}`);
    return this.finish(started.id, { status: 'SUCCEEDED', result });
  }

  /** The newest runs, of one job or of all. */
  async recent(limit: number, job?: JobName): Promise<JobRun[]> {
    if (!Number.isInteger(limit) || limit < 1) {
      throw new RangeError(`Limit must be a positive integer, got ${limit}`);
    }
    return this.prisma.jobRun.findMany({
      where: job ? { job } : {},
      orderBy: NEWEST_FIRST,
      take: limit,
    });
  }

  /** The newest run of every job, null for a job that never ran. */
  async latestPerJob(): Promise<Record<JobName, JobRun | null>> {
    const latest = await Promise.all(
      JOB_NAMES.map((job) =>
        this.prisma.jobRun.findFirst({ where: { job }, orderBy: NEWEST_FIRST }),
      ),
    );
    return Object.fromEntries(
      JOB_NAMES.map((job, index) => [job, latest[index]]),
    ) as Record<JobName, JobRun | null>;
  }

  /**
   * Marks runs that are RUNNING for longer than `staleHours` as FAILED:
   * their pod ended without a result (activeDeadlineSeconds, OOM, node
   * loss). finishedAt is when this was noticed. Each affected job gets the
   * usual failure notification - otherwise a job that is killed every time
   * would fail without anyone hearing about it. Compared against now(), the
   * database clock. Returns how many runs were aborted.
   */
  async abortStale(staleHours: number): Promise<number> {
    const aborted = await this.prisma.$queryRaw<{ id: string; job: string }[]>`
      UPDATE "JobRun"
      SET "status" = 'FAILED', "finishedAt" = now(), "error" = ${JOB_ABORTED_ERROR}
      WHERE "status" = 'RUNNING'
        AND "startedAt" < now() - (${staleHours} * interval '1 hour')
      RETURNING "id", "job"`;
    const firstRunPerJob = new Map<JobName, string>();
    for (const { id, job } of aborted) {
      if (isJobName(job) && !firstRunPerJob.has(job)) {
        firstRunPerJob.set(job, id);
      }
    }
    for (const [job, runId] of firstRunPerJob) {
      await this.notifyFailure(job, runId);
    }
    return aborted.length;
  }

  private async fail(
    job: JobName,
    runId: string,
    error: unknown,
  ): Promise<JobRun> {
    const { message, stack } = loggableError(error);
    this.logger.error(`Job "${job}" failed: ${message}`, stack);
    const failed = await this.finish(runId, {
      status: 'FAILED',
      error: message,
    });
    await this.notifyFailure(job, runId);
    return failed;
  }

  /**
   * Ends a run. finishedAt is now(), the database clock - the same clock
   * abortStale and every cleanup comparison use.
   */
  private async finish(runId: string, outcome: RunOutcome): Promise<JobRun> {
    const result =
      outcome.status === 'SUCCEEDED' ? JSON.stringify(outcome.result) : null;
    const error = outcome.status === 'FAILED' ? outcome.error : null;
    const [finished] = await this.prisma.$queryRaw<JobRun[]>`
      UPDATE "JobRun"
      SET "status" = CAST(${outcome.status} AS "JobRunStatus"),
          "finishedAt" = now(),
          "result" = CAST(${result} AS jsonb),
          "error" = ${error}
      WHERE "id" = ${runId}
      RETURNING *`;
    if (!finished) {
      throw new Error(`JobRun ${runId} no longer exists`);
    }
    return finished;
  }

  /**
   * At most one notification per job and UTC day (database clock): the
   * hourly app-backup check would otherwise report the same outage every
   * hour. The advisory lock per job serialises check and write, so two
   * runs failing at once cannot both pass the check. A notification that
   * cannot be written is logged; the run stays FAILED either way.
   */
  private async notifyFailure(job: JobName, runId: string): Promise<void> {
    try {
      await this.prisma.$transaction(async (tx) => {
        await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`job-failure-notification:${job}`}, 0))`;
        const [{ notified }] = await tx.$queryRaw<{ notified: boolean }[]>`
          SELECT EXISTS (
            SELECT 1 FROM "Notification" n
            JOIN "JobRun" r ON r."id" = n."relatedId"
            WHERE n."relatedType" = ${JOB_RUN_RELATED_TYPE}
              AND r."job" = ${job}
              AND n."createdAt" >= date_trunc('day', now())
          ) AS "notified"`;
        if (notified) {
          return;
        }
        await this.notifications.recordForPermission(
          JOB_FAILURE_PERMISSION[job],
          {
            priority: 'WARNUNG',
            message: `Job „${JOB_LABELS_DE[job]}“ fehlgeschlagen`,
            relatedType: JOB_RUN_RELATED_TYPE,
            relatedId: runId,
          },
          tx,
        );
      });
    } catch (error: unknown) {
      const { message, stack } = loggableError(error);
      this.logger.error(
        `Failure notification for job "${job}" could not be written: ${message}`,
        stack,
      );
    }
  }
}