105 lines
3.1 KiB
TypeScript
105 lines
3.1 KiB
TypeScript
import { Inject, Injectable, NotFoundException } from '@nestjs/common';
|
|
import { ACTIVE_JOB_STATUSES, BackgroundJobRecord, EnqueueJobInput, JOBS_PRISMA, MarkFailedOptions } from './jobs.types';
|
|
|
|
const DEFAULT_MAX_ATTEMPTS = 3;
|
|
const DEFAULT_RETRY_DELAY_MS = 60_000;
|
|
const MAX_ERROR_LENGTH = 2000;
|
|
|
|
@Injectable()
|
|
export class JobsService {
|
|
constructor(@Inject(JOBS_PRISMA) private readonly prisma: any) {}
|
|
|
|
async enqueue(input: EnqueueJobInput): Promise<BackgroundJobRecord> {
|
|
const type = input.type.trim();
|
|
if (!type) throw new Error('Job type is required');
|
|
const dedupeKey = input.dedupeKey?.trim() || null;
|
|
|
|
if (dedupeKey) {
|
|
const existing = await this.findActiveDedupe(type, dedupeKey);
|
|
if (existing) return existing;
|
|
}
|
|
|
|
try {
|
|
return await this.delegate.create({
|
|
data: {
|
|
type,
|
|
dedupeKey,
|
|
payload: input.payload ?? {},
|
|
maxAttempts: input.maxAttempts ?? DEFAULT_MAX_ATTEMPTS,
|
|
availableAt: input.availableAt,
|
|
},
|
|
});
|
|
} catch (error) {
|
|
if (dedupeKey && isUniqueConflict(error)) {
|
|
const existing = await this.findActiveDedupe(type, dedupeKey);
|
|
if (existing) return existing;
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
markSucceeded(id: string, now = new Date()): Promise<BackgroundJobRecord> {
|
|
return this.delegate.update({
|
|
where: { id },
|
|
data: {
|
|
status: 'succeeded',
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
completedAt: now,
|
|
lastError: null,
|
|
},
|
|
});
|
|
}
|
|
|
|
async markFailed(id: string, error: unknown, options: MarkFailedOptions = {}): Promise<BackgroundJobRecord> {
|
|
const job = await this.delegate.findUnique({ where: { id } });
|
|
if (!job) throw new NotFoundException('Background job not found');
|
|
|
|
const now = options.now ?? new Date();
|
|
const attempts = Number(job.attempts ?? 0);
|
|
const maxAttempts = Number(job.maxAttempts ?? DEFAULT_MAX_ATTEMPTS);
|
|
const hasRetriesLeft = attempts < maxAttempts;
|
|
const availableAt = hasRetriesLeft
|
|
? new Date(now.getTime() + (options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS))
|
|
: now;
|
|
|
|
return this.delegate.update({
|
|
where: { id },
|
|
data: {
|
|
status: hasRetriesLeft ? 'queued' : 'failed',
|
|
availableAt,
|
|
lockedBy: null,
|
|
lockedUntil: null,
|
|
lastError: normalizeError(error),
|
|
},
|
|
});
|
|
}
|
|
|
|
private get delegate(): any {
|
|
return (this.prisma as any).backgroundJob;
|
|
}
|
|
|
|
private findActiveDedupe(type: string, dedupeKey: string): Promise<BackgroundJobRecord | null> {
|
|
return this.delegate.findFirst({
|
|
where: {
|
|
type,
|
|
dedupeKey,
|
|
status: { in: [...ACTIVE_JOB_STATUSES] },
|
|
},
|
|
orderBy: { createdAt: 'asc' },
|
|
});
|
|
}
|
|
}
|
|
|
|
function normalizeError(error: unknown): string {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
return message.slice(0, MAX_ERROR_LENGTH);
|
|
}
|
|
|
|
function isUniqueConflict(error: unknown): boolean {
|
|
return typeof error === 'object'
|
|
&& error !== null
|
|
&& 'code' in error
|
|
&& (error as { code?: unknown }).code === 'P2002';
|
|
}
|