defineJob

Defines a new job definition with typed input, output, and run function.

Signature

import { defineJob } from '@coji/durably'

const jobDef = defineJob<TName, TInput, TOutput>({
  name: TName,
  input: z.ZodType<TInput>,
  output?: z.ZodType<TOutput>,
  run: (step: StepContext, input: TInput) => Promise<TOutput>
})

Options

interface DefineJobConfig<TName, TInput, TOutput> {
  name: TName
  input: z.ZodType<TInput>
  output?: z.ZodType<TOutput>
  run: (step: StepContext, input: TInput) => Promise<TOutput>
}

Option	Type	Required	Description
name	string	Yes	Unique identifier for the job
input	ZodSchema	Yes	Zod schema for validating job input
output	ZodSchema	No	Zod schema for validating job output
run	Function	Yes	The job's run function

Run Function

The run function receives:

• step: The Step object for creating steps and logging
• input: The validated input

Returns

Returns a JobDefinition object that can be registered with durably.register().

Registering Jobs

Use durably.register() to register job definitions and get job handles:

const { job } = durably.register({
  job: jobDef,
})

// Multiple jobs at once
const { syncUsers, importCsv } = durably.register({
  syncUsers: syncUsersJob,
  importCsv: importCsvJob,
})

The job handle provides the following methods:

trigger()

await job.trigger(
  input: TInput,
  options?: TriggerOptions
): Promise<TriggerResult<TOutput>>

Triggers a new job run. Returns a TriggerResult which extends TypedRun with a disposition field indicating how the trigger was resolved ('created', 'idempotent', or 'coalesced').

Trigger Options

interface TriggerOptions<TLabels = Record<string, string>> {
  idempotencyKey?: string
  concurrencyKey?: string
  coalesce?: 'skip' | 'queue' | 'active'
  labels?: TLabels
}

interface TriggerAndWaitOptions<
  TLabels = Record<string, string>,
> extends TriggerOptions<TLabels> {
  timeout?: number
  pollingIntervalMs?: number
  onProgress?: (progress: ProgressData) => void | Promise<void>
  onLog?: (log: LogData) => void | Promise<void>
}

Option	Description
idempotencyKey	Prevents duplicate runs with the same key
concurrencyKey	Groups jobs for concurrency control (max 1 pending per key)
coalesce	Requires concurrencyKey. 'skip' and 'queue' are equivalent aliases: reuse an existing pending run or create one trailing pending run behind a leased run. 'active' reuses the oldest pending run, otherwise a same-job leased run with a non-expired lease, otherwise a waiting run, otherwise creates a pending run. Idempotency is resolved first. Reused runs keep their original input, labels, and idempotency key. Terminal runs and null or expired leases do not block active creation. An active selection is valid at the database decision point; a returned leased run may become terminal before it is observed.
labels	Key/value labels for filtering (type-safe when schema provided)
timeout	Timeout in ms (triggerAndWait only)
pollingIntervalMs	Storage poll interval while waiting (triggerAndWait only); inherits createDurably({ pollingIntervalMs }) when omitted — same behavior as waitForRun
onProgress	Progress callback (triggerAndWait only)
onLog	Log callback (triggerAndWait only)

triggerAndWait()

await job.triggerAndWait(
  input: TInput,
  options?: TriggerAndWaitOptions
): Promise<{ id: string; output: TOutput; disposition: Disposition }>

Triggers a run and waits for completion. Throws if the run fails.

const { id, output, disposition } = await job.triggerAndWait({
  orgId: 'org_123',
})
console.log('Completed:', output)

// With timeout
const { output } = await job.triggerAndWait(
  { orgId: 'org_123' },
  { timeout: 30000 }, // 30 seconds
)

triggerAndWait vs waitForRun

triggerAndWait() creates a new run and waits. To wait for an existing run, use durably.waitForRun(runId).

// Trigger first, wait later
const run = await job.trigger({ orgId: 'org_123' })
const completed = await durably.waitForRun(run.id)

batchTrigger()

await job.batchTrigger(
  inputs: (TInput | { input: TInput; options?: TriggerOptions })[]
): Promise<TriggerResult<TOutput>[]>

Triggers multiple runs. All inputs are validated before any runs are created.

// Simple batch
const runs = await job.batchTrigger([
  { orgId: 'org_1' },
  { orgId: 'org_2' },
  { orgId: 'org_3' },
])

// With per-item options
const runs = await job.batchTrigger([
  { input: { orgId: 'org_1' }, options: { idempotencyKey: 'key-1' } },
  { input: { orgId: 'org_2' }, options: { idempotencyKey: 'key-2' } },
])

getRun()

await job.getRun(id: string): Promise<Run<TOutput> | null>

Gets a run by ID (only returns runs for this job).

getRuns()

await job.getRuns(filter?: { status?, limit?, offset? }): Promise<Run<TOutput>[]>

Gets runs for this job with optional filtering.

Example

import { createDurably, defineJob } from '@coji/durably'
import { z } from 'zod'

// Define the job
const syncUsersJob = defineJob({
  name: 'sync-users',
  input: z.object({
    orgId: z.string(),
    force: z.boolean().optional(),
  }),
  output: z.object({
    syncedCount: z.number(),
    errors: z.array(z.string()),
  }),
  run: async (step, input) => {
    const users = await step.run('fetch-users', async () => {
      return await api.fetchUsers(input.orgId)
    })

    const errors: string[] = []
    for (const user of users) {
      await step.run(`sync-${user.id}`, async () => {
        try {
          await db.upsertUser(user)
        } catch (e) {
          errors.push(`Failed to sync ${user.id}`)
        }
      })
    }

    return {
      syncedCount: users.length - errors.length,
      errors,
    }
  },
})

// Register with durably instance
const { syncUsers } = durably.register({
  syncUsers: syncUsersJob,
})

// Trigger the job
await syncUsers.trigger({ orgId: 'org_123' })

// With idempotency
await syncUsers.trigger(
  { orgId: 'org_123' },
  { idempotencyKey: 'sync-org_123-2024-01-01' },
)

Type Inference

Input and output types are inferred from the Zod schemas:

const exampleJob = defineJob({
  name: 'example',
  input: z.object({ id: z.string() }),
  output: z.object({ result: z.number() }),
  run: async (step, input) => {
    // input is typed as { id: string }
    return { result: 42 } // Must match output schema
  },
})

const { job } = durably.register({
  job: exampleJob,
})

// trigger() is typed
await job.trigger({ id: 'abc' }) // OK
await job.trigger({ wrong: 1 }) // Type error

Idempotent Registration

Registering the same JobDefinition instance multiple times returns the same job handle:

const jobDef = defineJob({ name: 'my-job', ... })

const { job: handle1 } = durably.register({ job: jobDef })
const { job: handle2 } = durably.register({ job: jobDef })

console.log(handle1 === handle2) // true

This enables safe usage in React components where effects may run multiple times.