메뉴
Create and manage background jobs with graphile-worker using makeJob and makeCronJob. Use when adding new background jobs, cron jobs, fan-out patterns, enqueuing work, or working with the worker infrastructure.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/seasonedcc/seasoned-skills --skill background-jobs명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
SKILL.md 표시 중
| name | background-jobs |
| description | Create and manage background jobs with graphile-worker using makeJob and makeCronJob. Use when adding new background jobs, cron jobs, fan-out patterns, enqueuing work, or working with the worker infrastructure. |
Background jobs use graphile-worker via makeJob and makeCronJob from ~/framework/worker.server. The app registers its jobs in app/business/jobs.server.ts and runs them in a separate process (run-worker.ts).
graphile-worker mutates its own graphile_worker schema internally — that is fine; it is third-party-owned. Job handlers themselves write to the application schema, so they follow {{write-discipline}} (load the database-design skill).
makeJob(jobName, run, options?) — On-Demand JobCreates a job that is enqueued explicitly from business logic.
import { makeJob } from '~/framework/worker.server'
const processFile = makeJob(
'processFile',
async ({ fileId }: { fileId: string }) => {
// job logic
},
{ priority: 5 },
)
Returns { enqueue, run, jobName }:
enqueue(payload) — adds the job to the queue (uses the priority from the job definition)run(payload, helpers) — the handler function (also used directly in tests)The optional options parameter accepts { priority?: number }. Lower number = higher priority (runs first). Default is 0 (from graphile-worker).
makeCronJob(jobName, cronMatch, run, options?) — Scheduled JobCreates a job that runs on a cron schedule.
import { makeCronJob } from '~/framework/worker.server'
const cleanupExpired = makeCronJob(
'cleanupExpired',
'0 * * * *', // every hour
async () => {
// job logic
},
{ priority: 13 },
)
Returns everything from makeJob plus cronItem with the schedule. The options parameter works the same as makeJob — priority is applied both to the cron item and to any manual enqueue calls.
Job names must start with a verb. Both the variable name and the job name string (first argument to makeJob/makeCronJob) must match.
// Good: verb-first names
const sendAuthEmail = makeJob('sendAuthEmail', ...)
const processDocumentIntake = makeJob('processDocumentIntake', ...)
const pollDocumentIntake = makeCronJob('pollDocumentIntake', ...)
// Bad: noun-first names
const documentIntakeProcess = makeJob('documentIntakeProcess', ...)
const documentIntakePoll = makeCronJob('documentIntakePoll', ...)
Every job (both makeJob and makeCronJob) must be added to the jobs array in app/business/jobs.server.ts:
import { sendAuthEmail } from './auth.server'
import { myNewJob, myNewCronJob } from './my-feature.server'
const jobs = [sendAuthEmail, myNewJob, myNewCronJob]
export { jobs }
The worker runner uses this array to build both the task list and cron schedule.
For jobs triggered by business logic (e.g., sending emails after an action).
Define the job in a business file and call .enqueue() from other business functions:
const sendNotification = makeJob(
'sendNotification',
async ({ userId, message }: { userId: string; message: string }) => {
const user = await db()
.selectFrom('users')
.select('email')
.where('id', '=', userId)
.executeTakeFirst()
if (!user) return
// send email...
},
)
// Called from another function:
await sendNotification.enqueue({ userId: user.id, message: 'Welcome!' })
For processing multiple items in parallel. A cron job discovers work and enqueues individual child jobs.
Parent cron job — discovers items, creates records in a transaction, then enqueues after commit:
const enqueuePendingItems = makeCronJob(
'enqueuePendingItems',
'*/5 * * * *',
async () => {
const pending = await db()
.selectFrom('items')
.select('id')
.where(({ not, exists, selectFrom }) =>
not(
exists(
selectFrom('records')
.select('id')
.whereRef('records.itemId', '=', 'items.id'),
),
),
)
.execute()
await Promise.all(
pending.map(async (item) => {
const intake = await db()
.transaction()
.execute(async (trx) => {
const record = await trx
.insertInto('records')
.values({ itemId: item.id })
.returning('id')
.executeTakeFirstOrThrow()
return await trx
.insertInto('intakes')
.values({ recordId: record.id })
.returning('id')
.executeTakeFirstOrThrow()
})
await processItem.enqueue({ itemId: item.id, intakeId: intake.id })
}),
)
},
)
Child job — processes a single item:
const processItem = makeJob(
'processItem',
async ({ itemId, intakeId }: { itemId: string; intakeId: string }) => {
// Idempotency check first: query the database for evidence the work
// already happened, and return early when it did (see Idempotency below)
// process...
},
)
Both jobs must be registered in jobs.server.ts.
Track which step failed by using a mutable step variable. Wrap the success path in a transaction (see Transactions section below) and record failures outside the transaction so they always persist. Always re-throw the error after recording it — graphile-worker only retries jobs that throw, so swallowing errors silently marks the job as succeeded:
const processFile = makeJob(
'processFile',
async ({ fileId, intakeId }: { fileId: string; intakeId: string }) => {
let step = 'download'
try {
await db()
.transaction()
.execute(async (trx) => {
const existing = await trx
.selectFrom('successes')
.select('id')
.where('intakeId', '=', intakeId)
.executeTakeFirst()
if (existing) return
const data = await downloadFile(fileId)
step = 'upload'
await uploadToS3(key, data)
await trx
.insertInto('successes')
.values({ intakeId })
.execute()
})
} catch (error) {
const errorMessage =
error instanceof Error ? error.message : String(error)
await db()
.insertInto('failures')
.values({ intakeId, step, errorMessage })
.execute()
throw error
}
},
)
When a job runs multiple database queries, wrap them in a transaction for atomicity:
await db()
.transaction()
.execute(async (trx) => {
// Use trx instead of db() for all queries inside the callback
const record = await trx.selectFrom('items').select('id').where('id', '=', itemId).executeTakeFirstOrThrow()
await trx.insertInto('itemCompletions').values({ itemId: record.id }).execute()
})
Rules:
trx (not db()) for all queries inside the transaction callback// Good: minimal payload with only what's needed
const processItem = makeJob(
'processItem',
async ({ itemId, intakeId }: { itemId: string; intakeId: string }) => {
// ...
},
)
// Bad: including extra data not used by the job
const processItem = makeJob(
'processItem',
async ({ itemId, intakeId, originalName, createdBy }: { ... }) => {
// originalName and createdBy are never used
},
)
Jobs may be retried by graphile-worker. Always check if the work has already been done before doing it: derive "already done" from the database itself — query for the evidence the work leaves behind, keyed by the payload's ids — and return early when it is found. The check runs as the handler's first step, before any side effect.
For fan-out patterns, the parent cron job should also prevent duplicate enqueuing — typically by checking a database table or using unique constraints.
Call .run() directly with the payload and an empty JobHelpers:
import type { JobHelpers } from 'graphile-worker'
it('processes the item', async () => {
await myJob.run(
{ itemId: 'test-id' },
{} as JobHelpers,
)
// assert results...
})
Spy on .enqueue() and mock its return value:
it('enqueues the job', async () => {
const spy = vi.spyOn(myJob, 'enqueue').mockResolvedValue()
await functionThatEnqueues()
expect(spy).toHaveBeenCalledOnce()
expect(spy).toHaveBeenCalledWith({ itemId: 'expected-id' })
})
Assert that errors are both recorded to the failure table and re-thrown using rejects.toThrow():
it('records failure and re-throws when download fails', async () => {
const { intakeId } = await createIntakeRecords(fileId, filename)
mockedDownloadFile.mockRejectedValue(new Error('download failed'))
await expect(
processItem.run(
{ fileId, intakeId },
{} as JobHelpers,
),
).rejects.toThrow('download failed')
const failure = await db()
.selectFrom('failures')
.selectAll()
.where('intakeId', '=', intakeId)
.executeTakeFirstOrThrow()
expect(failure.step).toBe('download')
expect(failure.errorMessage).toBe('download failed')
})
Spy on the child job's .enqueue() in beforeEach, then call the parent function:
describe('pollAndEnqueue', () => {
let enqueueSpy: ReturnType<typeof vi.spyOn>
beforeEach(() => {
enqueueSpy = vi
.spyOn(childJob, 'enqueue')
.mockResolvedValue()
})
it('enqueues for each pending item', async () => {
// set up test data...
await pollFunction()
expect(enqueueSpy).toHaveBeenCalledTimes(expectedCount)
})
})
The worker runs as a separate process via run-worker.ts (at the repo root):
CronJob entriesapp/framework/worker.server.tsapp/business/jobs.server.tsrun-worker.tsapp/business/auth.server.tsx (sendAuthEmail)