ApiCore/ApiCore-Backend/test/jobs.e2e-spec.ts

import { INestApplicationContext } from '@nestjs/common';
import { Test } from '@nestjs/testing';
import { DailyEvaluationJob } from '../src/jobs/daily-evaluation.job';
import { JOB_ABORTED_ERROR } from '../src/system/job-run.service';
import { JobsModule } from '../src/jobs/jobs.module';
import { runCli } from '../src/jobs/run-cli';
import { PrismaService } from '../src/prisma/prisma.service';

const HOUR_MS = 60 * 60 * 1000;
const DAY_MS = 24 * HOUR_MS;

const ago = (ms: number): Date => new Date(Date.now() - ms);

/**
 * The jobs run like `node dist/cli.js <job>` does: runCli with the real
 * JobsModule context (or, for the failure case, the same module with one
 * job replaced). Rows are tagged with this file's suffix and removed in
 * afterAll - the test database is shared by parallel e2e files.
 */
describe('Jobs (e2e)', () => {
  const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  const testStart = new Date(Date.now() - 1000);
  let prisma: PrismaService;
  const jobRunIds: string[] = [];
  const taskIds: string[] = [];
  let templateId: string | undefined;
  let subscriptionId: string | undefined;
  let adminId: string | undefined;

  const runsSinceStart = (job: string) =>
    prisma.jobRun.findMany({
      where: { job, startedAt: { gte: testStart } },
      orderBy: { startedAt: 'asc' },
    });

  beforeAll(async () => {
    prisma = new PrismaService();
    await prisma.$connect();
  });

  afterAll(async () => {
    const runs = await prisma.jobRun.findMany({
      where: { startedAt: { gte: testStart } },
      select: { id: true },
    });
    const runIds = [...jobRunIds, ...runs.map((run) => run.id)];
    await prisma.notification.deleteMany({
      where: { relatedType: 'JobRun', relatedId: { in: runIds } },
    });
    await prisma.jobRun.deleteMany({ where: { id: { in: runIds } } });
    if (templateId) {
      const generated = await prisma.task.findMany({
        where: { sourceTemplateId: templateId },
        select: { id: true },
      });
      taskIds.push(...generated.map((task) => task.id));
      await prisma.auditLog.deleteMany({
        where: { entityType: 'Task', entityId: { in: taskIds } },
      });
      await prisma.domainEvent.deleteMany({
        where: { aggregateType: 'Task', aggregateId: { in: taskIds } },
      });
      await prisma.task.deleteMany({ where: { id: { in: taskIds } } });
      await prisma.taskTemplate.deleteMany({ where: { id: templateId } });
    }
    await prisma.domainEvent.deleteMany({
      where: { aggregateId: `jobs-e2e-${suffix}` },
    });
    if (subscriptionId) {
      await prisma.webhookSubscription.deleteMany({
        where: { id: subscriptionId },
      });
    }
    await prisma.throttlerHit.deleteMany({
      where: { key: { startsWith: `jobs-e2e-${suffix}` } },
    });
    await prisma.idempotencyRecord.deleteMany({
      where: { key: { startsWith: `jobs-e2e-${suffix}` } },
    });
    await prisma.consumedToken.deleteMany({
      where: { jti: { startsWith: `jobs-e2e-${suffix}` } },
    });
    if (adminId) {
      await prisma.notification.deleteMany({ where: { userId: adminId } });
      await prisma.user.deleteMany({ where: { id: adminId } });
    }
    await prisma.$disconnect();
  });

  describe('cleanup', () => {
    const tag = (name: string): string => `jobs-e2e-${suffix}-${name}`;
    const ids: Record<string, string> = {};

    beforeAll(async () => {
      const event = async (
        name: string,
        status: 'DONE' | 'FAILED' | 'PENDING',
        age: number,
      ) => {
        const created = await prisma.domainEvent.create({
          data: {
            type: 'TaskDue',
            aggregateType: 'Task',
            aggregateId: `jobs-e2e-${suffix}`,
            payload: {},
            correlationId: tag(name),
            status,
            occurredAt: ago(age),
          },
        });
        ids[`event:${name}`] = created.id;
      };
      await event('done-old', 'DONE', 31 * DAY_MS);
      await event('done-recent', 'DONE', 29 * DAY_MS);
      await event('failed-old', 'FAILED', 91 * DAY_MS);
      await event('failed-recent', 'FAILED', 89 * DAY_MS);
      await event('pending-old', 'PENDING', 91 * DAY_MS);

      const subscription = await prisma.webhookSubscription.create({
        data: {
          url: 'https://receiver.example.com/hook',
          events: ['task.created'],
          secret: 'not-used',
          active: false,
        },
      });
      subscriptionId = subscription.id;
      const delivery = async (
        name: string,
        status: 'DELIVERED' | 'FAILED' | 'PENDING',
        age: number,
      ) => {
        const created = await prisma.webhookDelivery.create({
          data: {
            subscriptionId: subscription.id,
            event: 'task.created',
            body: '{}',
            status,
            createdAt: ago(age),
          },
        });
        ids[`delivery:${name}`] = created.id;
      };
      await delivery('delivered-old', 'DELIVERED', 31 * DAY_MS);
      await delivery('delivered-recent', 'DELIVERED', 29 * DAY_MS);
      await delivery('failed-old', 'FAILED', 91 * DAY_MS);
      await delivery('failed-recent', 'FAILED', 89 * DAY_MS);
      await delivery('pending-old', 'PENDING', 91 * DAY_MS);

      // "rekey" notifies holders of system:manage; cleanup sends that for
      // the stale run, the afterAll removes it by relatedId.
      const run = async (
        name: string,
        status: 'RUNNING' | 'SUCCEEDED' | 'FAILED',
        age: number,
      ) => {
        const created = await prisma.jobRun.create({
          data: { job: 'rekey', status, startedAt: ago(age) },
        });
        ids[`run:${name}`] = created.id;
        jobRunIds.push(created.id);
      };
      await run('succeeded-old', 'SUCCEEDED', 91 * DAY_MS);
      await run('failed-recent', 'FAILED', 89 * DAY_MS);
      await run('running-stale', 'RUNNING', 7 * HOUR_MS);
      await run('running-fresh', 'RUNNING', 5 * HOUR_MS);

      await prisma.throttlerHit.createMany({
        data: [
          {
            key: tag('expired'),
            throttlerName: 'default',
            hits: 3,
            expiresAt: ago(60_000),
          },
          {
            key: tag('blocked'),
            throttlerName: 'default',
            hits: 11,
            expiresAt: ago(60_000),
            blockedUntil: ago(-HOUR_MS),
          },
          {
            key: tag('block-ended'),
            throttlerName: 'default',
            hits: 11,
            expiresAt: ago(2 * 60_000),
            blockedUntil: ago(60_000),
          },
          {
            key: tag('open-window'),
            throttlerName: 'default',
            hits: 1,
            expiresAt: ago(-60_000),
          },
        ],
      });

      await prisma.idempotencyRecord.createMany({
        data: [
          {
            userId: tag('user'),
            key: tag('old'),
            scope: 'POST /api/v1/tasks',
            createdAt: ago(91 * DAY_MS),
          },
          {
            userId: tag('user'),
            key: tag('recent'),
            scope: 'POST /api/v1/tasks',
            createdAt: ago(89 * DAY_MS),
          },
        ],
      });
      await prisma.consumedToken.createMany({
        data: [
          { jti: tag('expired'), expiresAt: ago(2 * HOUR_MS) },
          // Expired, but within the 1 h margin for api pods' clocks.
          { jti: tag('in-grace'), expiresAt: ago(30 * 60_000) },
          { jti: tag('valid'), expiresAt: ago(-HOUR_MS) },
        ],
      });
    });

    it('node dist/cli.js cleanup equivalent: runCli(["cleanup"]) writes a SUCCEEDED JobRun with counts', async () => {
      expect(await runCli(['cleanup'])).toBe(0);

      const [run, ...others] = await runsSinceStart('cleanup');
      expect(others).toHaveLength(0);
      expect(run.status).toBe('SUCCEEDED');
      expect(run.finishedAt).toEqual(expect.any(Date));
      expect(run.error).toBeNull();
      const result = run.result as Record<string, number>;
      // Other e2e files share the database: at least this file's rows.
      expect(result.idempotencyRecords).toBeGreaterThanOrEqual(1);
      expect(result.consumedTokens).toBeGreaterThanOrEqual(1);
      expect(result.throttlerHits).toBeGreaterThanOrEqual(2);
      expect(result['domainEvents.done']).toBeGreaterThanOrEqual(1);
      expect(result['domainEvents.failed']).toBeGreaterThanOrEqual(1);
      expect(result['webhookDeliveries.delivered']).toBeGreaterThanOrEqual(1);
      expect(result['webhookDeliveries.failed']).toBeGreaterThanOrEqual(1);
      expect(result['jobRuns.aborted']).toBeGreaterThanOrEqual(1);
      expect(result['jobRuns.deleted']).toBeGreaterThanOrEqual(1);
    });

    it('removed exactly the expired rows', async () => {
      const remainingEvents = await prisma.domainEvent.findMany({
        where: { aggregateId: `jobs-e2e-${suffix}` },
        select: { correlationId: true },
      });
      expect(remainingEvents.map((e) => e.correlationId).sort()).toEqual(
        [tag('done-recent'), tag('failed-recent'), tag('pending-old')].sort(),
      );

      const remainingDeliveries = await prisma.webhookDelivery.findMany({
        where: { subscriptionId },
        select: { id: true },
      });
      expect(remainingDeliveries.map((d) => d.id).sort()).toEqual(
        [
          ids['delivery:delivered-recent'],
          ids['delivery:failed-recent'],
          ids['delivery:pending-old'],
        ].sort(),
      );

      const remainingHits = await prisma.throttlerHit.findMany({
        where: { key: { startsWith: `jobs-e2e-${suffix}` } },
        select: { key: true },
      });
      expect(remainingHits.map((h) => h.key).sort()).toEqual(
        [tag('blocked'), tag('open-window')].sort(),
      );

      const remainingRecords = await prisma.idempotencyRecord.findMany({
        where: { userId: tag('user') },
        select: { key: true },
      });
      expect(remainingRecords.map((r) => r.key)).toEqual([tag('recent')]);
      const remainingTokens = await prisma.consumedToken.findMany({
        where: { jti: { startsWith: `jobs-e2e-${suffix}` } },
        select: { jti: true },
      });
      expect(remainingTokens.map((t) => t.jti).sort()).toEqual(
        [tag('in-grace'), tag('valid')].sort(),
      );
    });

    it('marks runs older than 6 h that never finished as aborted and deletes runs older than 90 days', async () => {
      const runs = await prisma.jobRun.findMany({
        where: {
          id: {
            in: [
              ids['run:succeeded-old'],
              ids['run:failed-recent'],
              ids['run:running-stale'],
              ids['run:running-fresh'],
            ],
          },
        },
      });
      const byId = new Map(runs.map((run) => [run.id, run]));

      expect(byId.has(ids['run:succeeded-old'])).toBe(false);
      expect(byId.get(ids['run:failed-recent'])?.status).toBe('FAILED');
      expect(byId.get(ids['run:running-stale'])).toMatchObject({
        status: 'FAILED',
        error: JOB_ABORTED_ERROR,
        finishedAt: expect.any(Date) as Date,
      });
      expect(byId.get(ids['run:running-fresh'])).toMatchObject({
        status: 'RUNNING',
        finishedAt: null,
      });
    });
  });

  describe('task-templates', () => {
    it('task-templates creates the due task once even when run twice', async () => {
      const template = await prisma.taskTemplate.create({
        data: {
          title: `Jobs e2e ${suffix}`,
          recurrenceType: 'INTERVAL_DAYS',
          intervalDays: 7,
          nextDueDate: ago(DAY_MS),
        },
      });
      templateId = template.id;

      expect(await runCli(['task-templates'])).toBe(0);
      expect(await runCli(['task-templates'])).toBe(0);

      const generated = await prisma.task.findMany({
        where: { sourceTemplateId: template.id },
      });
      expect(generated).toHaveLength(1);
      expect(generated[0].createdByUserId).toBeNull();
      const advanced = await prisma.taskTemplate.findUniqueOrThrow({
        where: { id: template.id },
      });
      expect(advanced.nextDueDate.getTime()).toBeGreaterThan(Date.now());

      const runs = await runsSinceStart('task-templates');
      expect(runs.map((run) => run.status)).toEqual(['SUCCEEDED', 'SUCCEEDED']);
      for (const run of runs) {
        expect(run.result).toEqual({ created: expect.any(Number) as number });
      }
    });
  });

  describe('a failing job', () => {
    // The same module as dist/cli.js, with the daily evaluation replaced by
    // a job that always fails.
    const failingContext = async (): Promise<INestApplicationContext> => {
      const moduleRef = await Test.createTestingModule({
        imports: [JobsModule],
      })
        .overrideProvider(DailyEvaluationJob)
        .useValue({
          name: 'daily-evaluation',
          run: () => Promise.reject(new Error('weather service unreachable')),
        })
        .compile();
      return moduleRef.init();
    };

    beforeAll(async () => {
      const admin = await prisma.user.create({
        data: {
          email: `jobs-e2e-admin-${suffix}@example.com`,
          passwordHash: 'not-used',
          name: 'Jobs Admin',
          role: { connect: { key: 'ADMIN' } },
        },
      });
      adminId = admin.id;
      // The dedupe is per job and UTC day across the whole database: an
      // earlier failure notification of today (another run of this file,
      // a crashed run) would suppress the one this test expects.
      await prisma.$executeRaw`
        DELETE FROM "Notification" n
        USING "JobRun" r
        WHERE r."id" = n."relatedId"
          AND n."relatedType" = 'JobRun'
          AND r."job" = 'daily-evaluation'
          AND n."createdAt" >= date_trunc('day', now())`;
    });

    it('exits 1, stores the error and notifies permission holders once per day', async () => {
      expect(await runCli(['daily-evaluation'], failingContext)).toBe(1);
      expect(await runCli(['daily-evaluation'], failingContext)).toBe(1);

      const runs = await runsSinceStart('daily-evaluation');
      expect(runs).toHaveLength(2);
      for (const run of runs) {
        expect(run).toMatchObject({
          status: 'FAILED',
          error: 'weather service unreachable',
          result: null,
        });
      }
      const notifications = await prisma.notification.findMany({
        where: { userId: adminId, relatedType: 'JobRun' },
      });
      expect(notifications).toEqual([
        expect.objectContaining({
          priority: 'WARNUNG',
          message: 'Job „Tägliche Auswertung“ fehlgeschlagen',
          relatedId: runs[0].id,
          read: false,
        }),
      ]);
    });
  });
});