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 ` 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 () => { // Only the jobs this file runs: other files (rekey, app-backup, voice) // record their own runs in the same window and read them back. const runs = await prisma.jobRun.findMany({ where: { job: { in: ['cleanup', 'daily-evaluation', 'task-templates'] }, 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 = {}; 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; // 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 => { 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, }), ]); }); }); });