Zero runtime dependencies

Queues without ceremony.

One small, fully inferred API for background work. Start in memory, switch to Redis without rewriting your jobs, and use it naturally from Node.js, Bun, or Hono.

typed by inference memory + redis 39 tests · Node + Bun cron schedules web standards

$ pnpm add enqiu

01 · Quick start

Your handlers are the types.

Define each payload once, next to the code that uses it. Enqiu infers every job name, input, returned result, and event shape. There is no job-map interface and no generic to pass.

import { enqiu, job } from 'enqiu'
import { z } from 'zod'

const jobs = enqiu({
  sendEmail: job({
    input: z.object({
      to: z.string().email(),
      subject: z.string(),
    }),
    run: async (input) => {
      // input is inferred and validated
      return { deliveredAt: Date.now() }
    },
  }),
})

const email = await jobs.sendEmail({
  to: 'hello@enqiu.dev',
  subject: 'Welcome',
})

// The selected driver has accepted the job.
const result = await email.result
//    ^? { deliveredAt: number }

import { enqiu } from 'enqiu'

const jobs = enqiu({
  sendEmail: async (
    input: { to: string; subject: string },
  ) => {
    return { deliveredAt: Date.now() }
  },
})

const email = await jobs.sendEmail({
  to: 'hello@enqiu.dev',
  subject: 'Welcome',
})

// No JobMap, interfaces, result types, or generics.
const result = await email.result
//    ^? { deliveredAt: number }

✓

The job name is never repeated

Once sendEmail is defined, it becomes a typed method. There are no string names at call sites. With Redis, the definition key is also its durable identity, so drain or migrate queued work before renaming it.

✓

No handwritten TypeScript types with schema-first jobs

Any Standard Schema-compatible library can provide both inference and runtime validation. Zod is shown only as an example and is not an Enqiu dependency.

02 · Type inference

One source of truth.

Types travel from the job definition through the entire lifecycle. Rename a job or change its payload and every caller updates with it.

1 Define

The handler or schema owns the input and result shape.

2 Call

Autocomplete knows valid names, payloads, and options.

3 Result

The job’s result promise has the handler’s exact return type.

inference.ts

await jobs.sendEmail({
  to: 42,
  //  ~~ Type 'number' is not assignable to type 'string'
})

await jobs.missingJob({})
//   ~~~~~~~~~~ Property 'missingJob' does not exist

jobs.queue.on('succeeded', (job) => {
  if (job.name === 'sendEmail') {
    job.output.deliveredAt
    //         ^? number — narrowed by job name
  }
})

03 · Drivers

Start small. Scale without a rewrite.

Memory is the default and needs no setup. Redis is one option away, while handlers and producer code stay exactly the same.

mem In memory Zero dependencies · tests · local work · single process

R Redis Durable · distributed · Node.js and Bun clients

// Memory is intentionally boring.
const jobs = enqiu(definitions)

await jobs.sendEmail({ to, subject })
// The in-process worker starts automatically.

import { RedisClient } from 'bun'
import { enqiu, redis } from 'enqiu'

const client = new RedisClient(Bun.env.REDIS_URL)
await client.connect()

const jobs = enqiu(definitions, {
  name: 'emails',
  driver: redis(client),
  worker: false, // Producer-only API process
})

import { createClient } from 'redis'
import { enqiu, redis } from 'enqiu'

const client = createClient({
  url: process.env.REDIS_URL,
})
await client.connect()

const jobs = enqiu(definitions, {
  name: 'emails',
  driver: redis(client),
  worker: false, // Producer-only API process
})

Redis workers are explicit in production: worker: { concurrency: 10 }. Set worker: false in an HTTP producer so importing job definitions never starts background work unexpectedly.

04 · Hono

Web-standard by design.

Enqiu uses platform APIs such as AbortSignal and Promise. Hono apps running on Node.js or Bun can validate a request with the same Standard Schema used by the job.

routes.ts

import { Hono } from 'hono'
import { sValidator } from '@hono/standard-validator'
import { jobs } from './jobs'

const app = new Hono()

app.post(
  '/emails',
  sValidator('json', jobs.sendEmail.input),
  async (c) => {
    const job = await jobs.sendEmail(
      c.req.valid('json'),
    )
    return c.json({ id: job.id }, 202)
  },
)

app.get('/jobs/:id', async (c) => {
  const job = await jobs.queue.get(c.req.param('id'))
  return job
    ? c.json(job)
    : c.json({ error: 'Not found' }, 404)
})

05 · Job policies

Advanced when the work demands it.

Policies sit beside the handler and inherit its input type. They do not add new producer methods or weaken inference, and none are required for an ordinary job.

sendEmail: job({
  input: emailSchema,
  expiresIn: 10 * 60_000,

  concurrency: {
    limit: 2,
    by: (input) => input.organizationId,
  },

  throttle: {
    limit: 100,
    per: 60_000,
    burst: 20,
  },

  run: sendEmail,
})

rebuildSearch: job({
  input: rebuildSchema,

  debounce: {
    by: (input) => input.organizationId,
    wait: 2_000,
    mode: 'trailing',
  },

  run: rebuildSearch,
})

// Rapid calls for one organization become one job.
// The latest valid input is used after two quiet seconds.

import { trace } from '@opentelemetry/api'

const tracer = trace.getTracer('enqiu')
const jobs = enqiu(definitions, {
  telemetry: {
    emit(event) {
      tracer.startSpan(event.type, {
        attributes: {
          'enqiu.queue': event.queue,
          'enqiu.job.id': event.job?.id,
        },
      }).end()
    },
  },
})

async function sendEmail(input, context) {
  context.log.info('Sending email', {
    organizationId: input.organizationId,
  })

  await context.reportProgress({
    completed: 1,
    total: 3,
  })
}

Enqiu accepts durations as milliseconds. Use plain numbers or an application-level helper such as ms(); it is not an Enqiu dependency.

debounce Collapse rapid duplicate work

leading runs the first call immediately; trailing waits and runs the latest call after the quiet period.

throttle Protect downstream services

Run at a steady limit while burst permits a small immediate spike without removing the long-term cap.

concurrency.by Fairness per customer

Apply the limit separately to each organization, account, or other inferred key so one tenant cannot occupy every worker.

expiresIn Drop stale work

Expire a job before it begins when executing it late would be incorrect. Expired jobs remain inspectable as a terminal state.

jobs.queue.pause() Redis-wide operations

Pause, resume, or change concurrency across every worker. Local process controls remain under jobs.worker.

context.log Observability without lock-in

Structured job logs and lifecycle events work alone. The small telemetry hook can forward events to OpenTelemetry without making tracing a required dependency.

06 · Cron schedules

Recurring work without another scheduler.

Schedule a defined job directly, so its name, input, and result stay inferred. Registering the same schedule again updates it instead of creating a duplicate.

schedules.ts

const schedule = await jobs.sendDigest.schedule({
  cron: '0 9 * * 1-5',
  timezone: 'Europe/Nicosia',
  input: { audience: 'active-users' },
})

schedule.id
schedule.nextRunAt

await schedule.pause()
await schedule.resume()
await schedule.remove()

Predictable time

UTC is the default. An IANA timezone makes daylight-saving behavior explicit.

Ordinary job runs

Every occurrence gets the same retries, logs, events, progress, and inspection as a manually submitted job.

Safe registration

One default schedule is upserted per job. Add an optional id only when a job needs multiple schedules.

Honest durability

Redis schedules survive restarts and coordinate across workers. Memory schedules exist only for the current process lifetime.

Missed runs are skipped by default to avoid a restart storm. catchUp: true opts into replaying missed occurrences.

07 · Reliability

Useful power, quiet defaults.

The common path stays tiny. Production controls appear only when you need them. Memory and Redis share one behavioral contract, with durability differences stated plainly.

Retries + backoff

Fixed or exponential retries, configurable globally or per job.

Priority + delay

Schedule work or move urgent jobs ahead without changing handlers.

Idempotency + deduplication

A durable key returns the existing job, including during the configured completed-result retention window.

Cancellation

Remove waiting work or cooperatively abort a running handler through the standard AbortSignal API.

Crash recovery

Redis leases let another worker reclaim abandoned work.

Graceful shutdown

Stop accepting work and drain active handlers before exit.

Dead-letter + redrive

Exhausted jobs remain inspectable and can be retried after their underlying problem is fixed.

Structured progress + logs

Report completed and total units, attach JSON-safe fields, and inspect the retained execution story after a failure.

The reliability contract

at least once A worker crash can run a handler again. Side-effecting handlers should therefore be idempotent.

JSON-safe data Inputs, results, errors, progress, and log fields have the same serializable shape in memory and Redis.

best-effort cancel Waiting jobs stop immediately; running handlers must observe context.signal.

explicit retention Completed results, failed jobs, logs, and idempotency keys have separate configurable retention periods.

memory ≠ durable Memory is behavior-compatible but disappears with the process. Redis persists and coordinates work across machines.

08 · API surface

A small core with progressive power.

Call jobs directly. Definition policies describe behavior, queue methods affect distributed state, and worker methods affect only the current process.

enqiu(definitions, options?) Create a typed memory or Redis-backed queue.

await jobs.<name>(input, options?) Submit a job and receive its inferred handle after the selected driver accepts it.

await jobs.<name>.bulk(inputs) Add many inferred jobs without repeating the job name.

jobs.<name>.schedule(options) Upsert a typed cron schedule for the defined job.

jobs.queue.get(id) · list(query) · stats() Inspect jobs with status, job type, time, and cursor filters.

jobs.queue.on(event, listener) Observe distributed lifecycle events with typed job narrowing in Redis, and in-process events in memory.

jobs.queue.pause() · resume() Pause or resume the entire Redis queue.

jobs.queue.setConcurrency(limit) Set a queue-wide limit across every Redis worker.

jobs.queue.redrive(id) Move a dead-letter job back to waiting after inspection.

jobs.worker.pause() · resume() Control only the worker running in this process.

jobs.worker.onIdle() · close() Drain and close the local worker during tests or deployment.

The job handle

job.id · job.status Available after acceptance; status is a snapshot.

await job.result Wait for the inferred result, or reject when the job fails, expires, or is cancelled.

await job.cancel(reason?) Request cancellation with the reliability rules above.

await job.refresh() Fetch fresh distributed state; immediate in memory.

Per-call options

producer.ts

import ms from 'ms'

const email = await jobs.sendEmail(input, {
  priority: 'high',
  delay: ms('5s'),
  expiresIn: ms('10m'),
  idempotencyKey: `welcome:${user.id}`,
})

Start simple

Call the work. Enqiu queues it.

Install Enqiu and start with the memory driver. Your job definitions, direct calls, inferred results, and Hono routes stay the same when you add Redis later.

$ pnpm add enqiu

no generics no type map schema optional direct job methods explicit job.result memory default redis(client) cron schedules debounce + throttle tenant fairness job expiration telemetry hooks