@samveys/queue

# createQueue Function

Builds a queue over a driver. The worker loop lives here; drivers only store rows.

function createQueue(options: QueueOptions): Queue;

Exported from packages/queue/src/queue.ts

# createQueueCommands Function

The queue commands. `resolve` returns the queue bound in the app's container.

function createQueueCommands(resolve: (app: AppContext) => Queue): readonly QueueCommand[];

Exported from packages/queue/src/commands.ts

# createQueueTables Function

Creates the queue tables if they do not exist yet. Run it from a migration or at deploy time.

function createQueueTables(db: Database, jobs?: string, failed?: string): Promise<void>;

Exported from packages/queue/src/database-driver.ts

# databaseQueueDriver Function

The `database` queue driver. Reservation is race-free without an external lock: - PostgreSQL / MySQL / MariaDB: `SELECT … FOR UPDATE SKIP LOCKED` then `UPDATE`, in one transaction; - SQLite: one atomic `UPDATE … WHERE id = (SELECT … LIMIT 1) … RETURNING *`. Reserving counts an attempt, so a job whose worker crashed cannot loop forever.

function databaseQueueDriver(options: DatabaseDriverOptions): QueueDriver;

Exported from packages/queue/src/database-driver.ts

# defineJob Function

Defines a job: a named handler with a retry policy.

function defineJob<P>(input: JobInput<P>): Job<P>;

Exported from packages/queue/src/job.ts

# fake Function

A queue that records dispatches and runs nothing. Use it in tests: `queue.assertDispatched(job)`.

function fake(): FakeQueue;

Exported from packages/queue/src/fake.ts

# queue Function

The `@samveys/queue` plugin: binds `'queue'` and adds `queue:work`, `queue:failed` and `queue:retry`.

function queue(config?: QueueConfig): import("@samveys/core").Plugin<QueueConfig>;

Exported from packages/queue/src/plugin.ts

# queueTablesSql Function

DDL for the two queue tables on the database's dialect.

function queueTablesSql(db: Pick<Database, 'dialect'>, jobs?: string, failed?: string): readonly string[];

Exported from packages/queue/src/database-driver.ts

# DEFAULT_QUEUE Value

Default queue name used when a job or dispatch does not name one.

const DEFAULT_QUEUE = "default";

Exported from packages/queue/src/contracts.ts

# DatabaseDriverOptions Interface

No documentation comment.

interface DatabaseDriverOptions {
    /** Resolves the database lazily, so the driver can be configured before the database plugin boots. */
    readonly database: () => Database;

Exported from packages/queue/src/database-driver.ts

# DispatchOptions Interface

No documentation comment.

interface DispatchOptions {
    /** Do not make the job available before now + delayMs. */
    readonly delayMs?: number;
    /** Target queue; defaults to the job's queue. */
    readonly queue?: string;
    /**
     * Write the job row inside this database transaction (transactional outbox):
     * a rollback drops the job, a commit makes it durable.
     */
    readonly transaction?: QueueTransaction;
}

Exported from packages/queue/src/contracts.ts

# FailedJob Interface

A job that exhausted its attempts.

interface FailedJob {
    readonly id: string | number;
    readonly queue: string;
    readonly name: string;
    readonly payload: unknown;
    readonly exception: string;
    readonly failedAt: string;
}

Exported from packages/queue/src/contracts.ts

# FakeQueue Interface

Test double that records dispatches instead of storing them.

interface FakeQueue extends Queue {
    readonly dispatched: readonly {
        readonly job: Job<never>;
        readonly payload: unknown;
        readonly options: DispatchOptions;
    }[];
    assertDispatched<P = unknown>(job: Job<P> | string, predicate?: (payload: P) => boolean): void;

Exported from packages/queue/src/contracts.ts

# Job Interface

A defined job.

interface Job<P = unknown> {
    readonly kind: 'job';
    readonly name: string;
    readonly handler: JobHandler<P>;
    readonly retries: number;
    readonly backoff: readonly number[];
    readonly timeoutMs: number | undefined;
    readonly queue: string;
}

Exported from packages/queue/src/contracts.ts

# JobContext Interface

Context handed to a job handler on every attempt.

interface JobContext {
    /** 1-based attempt number. */
    readonly attempt: number;
    /** Row id of the job in the driver (null for the sync driver). */
    readonly jobId: string | number | null;
    /** Queue the job was reserved from. */
    readonly queue: string;
    /** Aborted when `timeoutMs` elapses. */
    readonly signal: AbortSignal;
}

Exported from packages/queue/src/contracts.ts

# JobInput Interface

Input accepted by `defineJob`.

interface JobInput<P> {
    readonly name: string;
    readonly handler: JobHandler<P>;
    /** Additional attempts after the first one. Default 3. */
    readonly retries?: number;
    /** Delay in ms before retry n (the last entry repeats). Default `[1000, 5000, 30000]`. */
    readonly backoff?: readonly number[];
    /** Abort and fail an attempt after this many ms. */
    readonly timeoutMs?: number;
    /** Default queue for this job. */
    readonly queue?: string;
}

Exported from packages/queue/src/contracts.ts

# PushInput Interface

No documentation comment.

interface PushInput {
    readonly name: string;
    readonly queue: string;
    readonly payload: unknown;
    readonly availableAt: number;
    readonly transaction?: QueueTransaction | undefined;
}

Exported from packages/queue/src/contracts.ts

# Queue Interface

No documentation comment.

interface Queue {
    /** Register jobs so workers can resolve them by name. */
    register(...jobs: readonly Job<never>[]): void;
    dispatch<P>(job: Job<P>, payload: P, options?: DispatchOptions): Promise<void>;
    /** Run the worker loop. */
    work(options?: WorkOptions): Promise<WorkResult>;
    failed(): Promise<readonly FailedJob[]>;
    retry(id: string | number | 'all'): Promise<number>;
    size(queue?: string): Promise<number>;
}

Exported from packages/queue/src/contracts.ts

# QueueCommand Interface

No documentation comment.

interface QueueCommand {
    readonly kind: 'command';
    readonly name: string;
    readonly description: string;
    readonly definition: {
        readonly description: string;
        readonly args?: Readonly<Record<string, ArgLike>>;
        readonly flags?: Readonly<Record<string, FlagLike>>;
        run(context: QueueCommandContext): Promise<number>;
    };
}

Exported from packages/queue/src/commands.ts

# QueueConfig Interface

Configuration of `queue()`.

interface QueueConfig {
    /**
     * `'database'` (default): jobs live in the `db` binding's `samveys_jobs` table.
     * `'sync'`: jobs run inline at dispatch (tests). Or pass any {@link QueueDriver}.
     */
    readonly driver?: 'database' | 'sync' | QueueDriver;
    /** Jobs workers may run, by name. */
    readonly jobs?: readonly Job<never>[];
    /** Database connection name for the `database` driver (default connection when omitted). */
    readonly connection?: string;
  …

interface QueueConfig {
    /**
     * `'database'` (default): jobs live in the `db` binding's `samveys_jobs` table.
     * `'sync'`: jobs run inline at dispatch (tests). Or pass any {@link QueueDriver}.
     */
    readonly driver?: 'database' | 'sync' | QueueDriver;
    /** Jobs workers may run, by name. */
    readonly jobs?: readonly Job<never>[];
    /** Database connection name for the `database` driver (default connection when omitted). */
    readonly connection?: string;
    readonly table?: string;
    readonly failedTable?: string;
    readonly visibilityMs?: number;
}

Exported from packages/queue/src/plugin.ts

# QueueDriver Interface

Storage backend for a queue.

interface QueueDriver {
    readonly name: string;
    push(input: PushInput): Promise<void>;
    /** Atomically reserve the next available job; two workers never get the same row. */
    reserve(queue: string, workerId: string, now: number): Promise<ReservedJob | null>;
    /** Delete a finished job. */
    complete(job: ReservedJob): Promise<void>;
    /** Put a job back with an incremented attempt count, available at `availableAt`. */
    release(job: ReservedJob, availableAt: number): Promise<void>;
    /** Move a job to the failed store. */
  …

interface QueueDriver {
    readonly name: string;
    push(input: PushInput): Promise<void>;
    /** Atomically reserve the next available job; two workers never get the same row. */
    reserve(queue: string, workerId: string, now: number): Promise<ReservedJob | null>;
    /** Delete a finished job. */
    complete(job: ReservedJob): Promise<void>;
    /** Put a job back with an incremented attempt count, available at `availableAt`. */
    release(job: ReservedJob, availableAt: number): Promise<void>;
    /** Move a job to the failed store. */
    fail(job: ReservedJob, error: Error): Promise<void>;
    failed(): Promise<readonly FailedJob[]>;
    /** Re-dispatch a failed job; returns false when the id is unknown. */
    retryFailed(id: string | number): Promise<boolean>;
    size(queue: string): Promise<number>;
}

Exported from packages/queue/src/contracts.ts

# QueueOptions Interface

No documentation comment.

interface QueueOptions {
    /** A storage driver, or `'sync'` to run every job inline at dispatch time (tests, local dev). */
    readonly driver: QueueDriver | 'sync';
    /** Jobs known to workers by name. Dispatching a job registers it as well. */
    readonly jobs?: readonly Job<never>[];
    /** Identifies this worker in `reserved_by`. */
    readonly workerId?: string;
    readonly now?: () => number;

Exported from packages/queue/src/queue.ts

# QueueTransaction Interface

Minimal structural view of a database transaction. Any `@samveys/database` transaction satisfies it; it is used for the transactional outbox.

interface QueueTransaction {
    rawQuery(sql: string, bindings?: readonly unknown[]): Promise<unknown>;
}

Exported from packages/queue/src/contracts.ts

# ReservedJob Interface

A job row as reserved by a driver.

interface ReservedJob {
    readonly id: string | number;
    readonly queue: string;
    readonly name: string;
    readonly payload: unknown;
    /** Attempts made before this reservation. */
    readonly attempts: number;
}

Exported from packages/queue/src/contracts.ts

# WorkOptions Interface

No documentation comment.

interface WorkOptions {
    readonly queue?: string;
    /** Process at most one job and return. */
    readonly once?: boolean;
    /** Stop after this many jobs. */
    readonly maxJobs?: number;
    /** Poll interval in ms when idle. Default 500. */
    readonly pollMs?: number;
    /** Aborting stops the loop after the current job. */
    readonly signal?: AbortSignal;
}

Exported from packages/queue/src/contracts.ts

# WorkResult Interface

No documentation comment.

interface WorkResult {
    readonly processed: number;
    readonly failed: number;
}

Exported from packages/queue/src/contracts.ts

# JobHandler Type

No documentation comment.

type JobHandler<P> = (payload: P, ctx: JobContext) => void | Promise<void>;

Exported from packages/queue/src/contracts.ts