Job Queue

The default execution engine stores background work in the commerce_jobs table and processes it with the built-in Postgres runner. Jobs are durable: they survive server restarts and can be retried on failure. Set jobs.adapter to replace the engine without changing task definitions.

With the built-in engine, jobs accumulate in pending status until the cron endpoint, server.runJobs(), or jobs.autorun drives a runner cycle. Push engines are driven by their native runtime instead.

Task definition

A task is a named handler with typed input, output, retry policy, and optional concurrency control:

import type { TaskDefinition } from "@porulle/core";

const sendEmailTask: TaskDefinition<
  { to: string; template: string; data: Record<string, unknown> },
  { messageId: string }
> = {
  slug: "email/send",
  handler: async ({ input, ctx }) => {
    await ctx.services.email.send({
      template: input.template,
      to: input.to,
      data: input.data,
    });
    return { output: { messageId: "ok" } };
  },
  retries: {
    attempts: 3,
    backoff: { type: "exponential", delay: 5000 },
  },
};

TaskDefinition fields

| Field | Type | Required | Description | | ------------- | ---------------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | | slug | string | Yes | Unique identifier. Convention: domain/action (e.g., email/send, webhooks/deliver) | | handler | (args) => Promise<{ output }> | Yes | Function that runs when the job is processed. Receives { input, ctx }. | | retries | { attempts, backoff? } | No | Retry policy. backoff.type is "fixed" or "exponential". backoff.delay is milliseconds. | | concurrency | { key, exclusive?, supersedes? } | No | key(input) identifies related jobs. exclusive defaults to true; supersedes replaces an older pending job for the same task, organization, and key. |

Handler context

| Field | Type | Description | | -------------- | ------------------ | ------------------------------------------------------ | | input | TInput | The job’s input payload (typed by the task definition) | | ctx.db | PluginDb | Database instance for direct queries | | ctx.logger | Logger | Structured logger (Pino) | | ctx.services | ServiceContainer | Kernel services (inventory, orders, catalog, etc.) |

Enqueueing jobs

Jobs are enqueued via the jobs adapter on the service container or from HookContext.jobs:

// From a hook handler
await context.jobs.enqueue(
  "email/send",
  {
    to: "customer@example.com",
    template: "order-confirmation",
    data: { orderNumber: "ORD-001", total: "$99.00" },
  },
  { organizationId: context.actor!.organizationId },
);

// With delayed execution
await context.jobs.enqueue(
  "appointment/reminder",
  { bookingId: "abc-123" },
  {
    organizationId: context.actor!.organizationId,
    delayMs: 24 * 60 * 60 * 1000,
    queue: "reminders",
  },
);

Enqueue options

| Option | Type | Default | Description | | ---------------- | --------- | ----------------------------- | ---------------------------------------------------------------- | | organizationId | string | — | Required tenant boundary for the job. Blank values are rejected. | | queue | string | "default" | Queue name. Runners can process specific queues. | | delayMs | number | — | Delay before the job becomes eligible. | | maxAttempts | number | Task policy or 1 | Total attempts, including the first execution. | | concurrencyKey | string | Task concurrency.key(input) | Overrides the task-derived key. | | supersedes | boolean | Task policy | Overrides pending-job supersession. |

Job lifecycle

pending ──→ processing ──→ succeeded
                │
                ├──→ pending (retry with backoff)
                │
                └──→ failed (max attempts reached)

| Status | Meaning | | ------------ | --------------------------------------------------------------- | | pending | Waiting to be claimed by a runner. Respects waitUntil if set. | | processing | Claimed by a runner, handler is executing | | succeeded | Handler returned successfully. Output stored. | | failed | Handler threw and max attempts reached. Error stored. |

Runner strategies

1. Built-in cron endpoint (serverless)

UC exposes GET /api/jobs/run automatically. Point your cron service at it:

{
  "crons": [
    { "path": "/api/jobs/run", "schedule": "* * * * *" },
    { "path": "/api/jobs/run?queue=reminders", "schedule": "*/5 * * * *" }
  ]
}

Query parameters:

| Param | Default | Description | | ------- | ----------- | ----------------------- | | queue | "default" | Which queue to process | | limit | 10 | Max jobs per invocation |

In production, the endpoint requires *:* permission. Pass an API key via x-api-key header.

2. In-process polling (long-running servers)

export default defineConfig({
  jobs: {
    autorun: {
      enabled: true,
      intervalMs: 10_000,
    },
    tasks: [sendEmailTask, processOrderTask],
  },
});

The runner uses SELECT FOR UPDATE SKIP LOCKED, so multiple instances can poll the same queue safely without double-processing.

For tasks with a concurrency key, the claim transaction excludes keys already in processing. If a claimed batch contains multiple exclusive jobs with the same key, only the oldest runs; the rest are released to pending without consuming an attempt. Jobs with different keys execute in parallel.

Processing order

The default claim order is oldest createdAt first. Configure a sortable field or a comparator when a queue needs a different fairness policy:

export default defineConfig({
  jobs: {
    processingOrder: { field: "createdAt", direction: "asc" },
    // Or: processingOrder: (left, right) =>
    //   Number(right.input.priority) - Number(left.input.priority),
  },
});

Sort specifications support createdAt, updatedAt, attempts, and taskSlug. A comparator receives the stored job records and must be deterministic.

3. Custom worker process

import { runPendingJobs } from "@porulle/core";

async function loop() {
  while (true) {
    const { processed } = await runPendingJobs({
      db: kernel.database.db,
      tasks: taskMap,
      queue: "default",
      limit: 20,
      logger: kernel.logger,
      services: kernel.services,
    });

    if (processed === 0) {
      await new Promise((r) => setTimeout(r, 5000));
    }
  }
}

loop();

runPendingJobs drives only the built-in Drizzle engine. A custom pull engine is polled through server.runJobs(). Push engines such as Inngest, Trigger.dev, and Cloudflare Workflows are invoked by their native runtime; do not enable jobs.autorun for them.

Retry and backoff

When a handler throws, the runner checks attempts < maxAttempts:

Under limit — Job returns to pending with a waitUntil computed from the backoff policy
At limit — Job marked failed with the error message stored

Attempt 1: immediate
Attempt 2: wait 5s   (exponential: 5000 × 2^0)
Attempt 3: wait 10s  (exponential: 5000 × 2^1)
Attempt 4: wait 20s  (exponential: 5000 × 2^2)
Attempt 5: failed

Database schema

CREATE TABLE commerce_jobs (
  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  organization_id TEXT NOT NULL,
  queue           TEXT NOT NULL DEFAULT 'default',
  task_slug       TEXT NOT NULL,
  status          TEXT NOT NULL DEFAULT 'pending',
  input           JSONB DEFAULT '{}',
  output          JSONB,
  error           TEXT,
  attempts        INTEGER NOT NULL DEFAULT 0,
  max_attempts    INTEGER NOT NULL DEFAULT 1,
  wait_until      TIMESTAMPTZ,
  concurrency_key TEXT,
  processing_started_at TIMESTAMPTZ,
  completed_at    TIMESTAMPTZ,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Monitoring queries

-- Pending jobs older than 1 hour (stuck)
SELECT id, task_slug, created_at, wait_until
FROM commerce_jobs
WHERE status = 'pending' AND created_at < NOW() - INTERVAL '1 hour';

-- Queue depth per queue
SELECT queue, COUNT(*) AS pending
FROM commerce_jobs WHERE status = 'pending'
GROUP BY queue;

-- Recent failures
SELECT id, task_slug, error, attempts, completed_at
FROM commerce_jobs WHERE status = 'failed'
ORDER BY completed_at DESC LIMIT 20;

-- Processing time per task
SELECT task_slug,
  COUNT(*) AS total,
  AVG(EXTRACT(EPOCH FROM (completed_at - processing_started_at))) AS avg_seconds
FROM commerce_jobs WHERE status = 'succeeded'
GROUP BY task_slug;

Register tasks

All tasks must be registered before jobs referencing them are processed. If a job’s task_slug has no matching task definition, the runner marks it as failed with “Unknown task slug”.

import { APPOINTMENT_EMAIL_TASKS } from "@porulle/plugin-appointments";

export default defineConfig({
  jobs: {
    tasks: [
      ...APPOINTMENT_EMAIL_TASKS,
      {
        slug: "analytics:daily-aggregate",
        handler: async ({ ctx }) => {
          // aggregate daily analytics
          return { output: { date: new Date().toISOString() } };
        },
      },
    ],
  },
});