58 lines
1.6 KiB
TypeScript
58 lines
1.6 KiB
TypeScript
import { Inject, Injectable } from '@nestjs/common';
|
|
import { BackgroundJobRecord, ClaimJobInput, JOBS_PRISMA } from './jobs.types';
|
|
|
|
const DEFAULT_LEASE_MS = 60_000;
|
|
|
|
@Injectable()
|
|
export class JobLockService {
|
|
constructor(@Inject(JOBS_PRISMA) private readonly prisma: any) {}
|
|
|
|
async claimNext(input: ClaimJobInput): Promise<BackgroundJobRecord | null> {
|
|
const now = input.now ?? new Date();
|
|
const leaseMs = input.leaseMs ?? DEFAULT_LEASE_MS;
|
|
const types = input.types?.map((type) => type.trim()).filter(Boolean) ?? [];
|
|
|
|
return (this.prisma as any).$transaction(async (tx: any) => {
|
|
const params: unknown[] = [now];
|
|
const typeCondition = buildTypeCondition(types, params);
|
|
const rows = await tx.$queryRawUnsafe(
|
|
`
|
|
SELECT id
|
|
FROM background_jobs
|
|
WHERE (
|
|
(status = 'queued' AND available_at <= $1)
|
|
OR (status = 'running' AND locked_until IS NOT NULL AND locked_until < $1)
|
|
)
|
|
${typeCondition}
|
|
ORDER BY available_at ASC, created_at ASC
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED
|
|
`,
|
|
...params,
|
|
);
|
|
const id = rows[0]?.id;
|
|
if (!id) return null;
|
|
|
|
return tx.backgroundJob.update({
|
|
where: { id },
|
|
data: {
|
|
status: 'running',
|
|
lockedBy: input.workerId,
|
|
lockedUntil: new Date(now.getTime() + leaseMs),
|
|
attempts: { increment: 1 },
|
|
lastError: null,
|
|
},
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
function buildTypeCondition(types: string[], params: unknown[]): string {
|
|
if (types.length === 0) return '';
|
|
const placeholders = types.map((type) => {
|
|
params.push(type);
|
|
return `$${params.length}`;
|
|
});
|
|
return `AND type IN (${placeholders.join(', ')})`;
|
|
}
|