dorfteich/apps/api/src/scheduler/scheduler.service.ts

import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { PinoLogger } from 'nestjs-pino';

import { ClockService } from '../common/clock.service';
import { AppConfig } from '../config/app-config.service';
import { PrismaService } from '../prisma/prisma.service';

export interface JobDefinition {
  /** Stable identity — also the `jobs` table primary key. */
  name: string;
  cadenceSeconds: number;
  run: () => Promise<void>;
}

/** How often the scheduler checks which registered jobs are due. */
const TICK_MS = 60_000;
/** A job stuck `RUNNING` past this (crashed process, never released) is
 * treated as available again rather than blocked forever. */
const STALE_LOCK_MS = 60 * 60_000;

/**
 * Generic maintenance-job scheduler (issue #31; data-model.md/`jobs`,
 * operations.md's job table). Every later maintenance job (version
 * thinning, compaction, quota reconciliation, orphan sweep, …) registers
 * here instead of growing its own timer loop.
 *
 * Due-ness and the run-mutex both live in the `jobs` table, not in
 * memory: `lastRunAt` is what makes the schedule survive an api restart,
 * and claiming a due job is a single atomic
 * `UPDATE ... WHERE status != 'RUNNING'` — the same "row as mutex"
 * technique works whether it's this process racing itself (two ticks
 * overlapping because a job ran long) or, defensively, two processes.
 */
@Injectable()
export class SchedulerService implements OnModuleInit, OnModuleDestroy {
  private readonly jobs = new Map<string, JobDefinition>();
  private timer: NodeJS.Timeout | undefined;

  constructor(
    private readonly prisma: PrismaService,
    private readonly clock: ClockService,
    private readonly config: AppConfig,
    private readonly logger: PinoLogger,
  ) {
    this.logger.setContext(SchedulerService.name);
  }

  register(job: JobDefinition): void {
    this.jobs.set(job.name, job);
  }

  /** Every job registered in this process (issue #86 admin panel). */
  definitions(): JobDefinition[] {
    return [...this.jobs.values()];
  }

  /**
   * Manual trigger from the admin panel (issue #86): runs `name` now,
   * regardless of cadence — only the run-mutex still applies, so a job
   * already running elsewhere reports `already_running` instead of
   * doubling up.
   */
  async runNow(name: string): Promise<'succeeded' | 'failed' | 'already_running'> {
    const job = this.jobs.get(name);
    if (!job) throw new Error(`unknown job: ${name}`);
    return this.claimAndRun(job, { force: true });
  }

  onModuleInit(): void {
    if (this.config.env.NODE_ENV === 'test') return; // tests drive jobs directly
    // A tick hitting a transient database failure (outage, or the backup
    // sidecar terminating connections mid-restore, #103) must retry on the
    // next tick, never crash the api via an unhandled rejection.
    this.timer = setInterval(
      () =>
        void this.tick().catch((error: unknown) => {
          this.logger.warn({ err: error }, 'scheduler tick failed; retrying on the next tick');
        }),
      TICK_MS,
    );
    this.timer.unref();
  }

  onModuleDestroy(): void {
    if (this.timer) clearInterval(this.timer);
  }

  /** One scheduling pass over every registered job; public for tests/manual triggers. */
  async tick(): Promise<void> {
    for (const job of this.jobs.values()) {
      await this.runIfDue(job);
    }
  }

  /** Runs `job` now if due and not already running elsewhere; a no-op otherwise. */
  async runIfDue(job: JobDefinition): Promise<void> {
    await this.claimAndRun(job, { force: false });
  }

  private async claimAndRun(
    job: JobDefinition,
    { force }: { force: boolean },
  ): Promise<'succeeded' | 'failed' | 'already_running'> {
    try {
      await this.prisma.job.upsert({
        where: { name: job.name },
        create: { name: job.name, cadenceSeconds: job.cadenceSeconds },
        update: {},
      });
    } catch (error) {
      // Two overlapping ticks can both try to create the row for a
      // never-seen-before job at once; whichever loses just means the row
      // already exists now, which is exactly what this call wants anyway.
      const isDuplicate =
        error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002';
      if (!isDuplicate) throw error;
    }

    const now = this.clock.now();
    const dueBefore = new Date(now.getTime() - job.cadenceSeconds * 1000);
    const staleLockBefore = new Date(now.getTime() - STALE_LOCK_MS);

    const claim = await this.prisma.job.updateMany({
      where: {
        name: job.name,
        AND: [
          // A manual trigger skips the due check, never the run-mutex.
          ...(force ? [] : [{ OR: [{ lastRunAt: null }, { lastRunAt: { lte: dueBefore } }] }]),
          { OR: [{ status: { not: 'RUNNING' } }, { lockedAt: { lte: staleLockBefore } }] },
        ],
      },
      data: { status: 'RUNNING', lockedAt: now, lastRunAt: now },
    });
    if (claim.count === 0) return 'already_running'; // or, unforced, simply not due

    try {
      await job.run();
      await this.prisma.job.update({
        where: { name: job.name },
        data: { status: 'IDLE', lastError: null, lastDurationMs: this.sinceMs(now) },
      });
      return 'succeeded';
    } catch (error) {
      const message = error instanceof Error ? error.message.slice(0, 500) : String(error);
      this.logger.error({ job: job.name, err: error }, 'maintenance job failed');
      await this.prisma.job.update({
        where: { name: job.name },
        data: { status: 'FAILED', lastError: message, lastDurationMs: this.sinceMs(now) },
      });
      return 'failed';
    }
  }

  private sinceMs(start: Date): number {
    return Math.max(0, this.clock.now().getTime() - start.getTime());
  }
}