### Install Peer Dependencies

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Install the necessary Prisma client and driver adapter for Prisma Queue.

```bash
pnpm add @prisma/client @prisma/adapter-pg
```

--------------------------------

### Install Prisma Queue

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Install the Prisma Queue package using npm or pnpm.

```bash
npm install @mgcrea/prisma-queue --save
# or
pnpm add @mgcrea/prisma-queue
```

--------------------------------

### Install Prisma Queue Dependencies

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Install the necessary packages for Prisma Queue and Prisma client.

```bash
npm install @mgcrea/prisma-queue
npm install --save-dev @prisma/client @prisma/adapter-pg
```

--------------------------------

### Complete Worker Usage Example

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Demonstrates a complete worker setup for an import queue using Prisma Queue. It includes job locking, data fetching, item processing with progress updates, cancellation handling, and final result reporting.

```typescript
import { createQueue, type PrismaJob } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma";

type ImportPayload = { sourceId: string; batch: number };
type ImportResult = { imported: number; failed: number };

const importQueue = createQueue<ImportPayload, ImportResult, typeof prisma>(
  {
    prisma,
    name: "import",
    maxConcurrency: 3,
    jobTimeout: 30 * 60 * 1000, // 30 minutes
  },
  async (job: PrismaJob<ImportPayload, ImportResult>, client) => {
    console.log(`Processing job ${job.id}: source=${job.payload.sourceId}`);

    // Check if another process is already handling this batch
    const locked = await job.isLocked();
    if (locked) {
      throw new Error("Batch is being processed elsewhere");
    }

    // Fetch fresh data from source
    const source = await client.importSource.findUniqueOrThrow({
      where: { id: job.payload.sourceId },
    });

    const items = await fetchBatch(source, job.payload.batch);
    let imported = 0;
    let failed = 0;

    // Process items, reporting progress
    for (let i = 0; i < items.length; i++) {
      // Check for cancellation
      if (job.signal.aborted) {
        throw new Error("Job was cancelled");
      }

      try {
        await client.record.create({ data: items[i] });
        imported++;
      } catch (error) {
        console.error(`Failed to import item ${i}:`, error);
        failed++;
      }

      // Update progress
      const percent = Math.round(((i + 1) / items.length) * 100);
      await job.progress(percent);
    }

    // Record final state
    await job.update({
      result: { imported, failed },
    });

    return { imported, failed };
  }
);

// Listen for completion
importQueue.on("success", (result, job) => {
  console.log(`Job ${job.id} imported: ${result.imported}, failed: ${result.failed}`);
});

// Start processing
await importQueue.start();
```

--------------------------------

### Configure PrismaClient with adapter for v2.0

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

When migrating to v2.0, you must configure your PrismaClient with a driver adapter like `@prisma/adapter-pg`. This example shows the setup required for Prisma 7+ compatibility.

```typescript
import { PrismaPg } from "@prisma/adapter-pg";
import { PrismaClient } from "./prisma/client/client.js";

const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL });
export const prisma = new PrismaClient({ adapter });
```

--------------------------------

### Create and Start a Basic Job Queue

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Set up a basic job queue for processing tasks like sending notifications. Ensure the queue is started to begin processing jobs.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "notifications",
    maxConcurrency: 10,
  },
  async (job, client) => {
    await sendNotification(job.payload);
    return { sent: true };
  }
);

await queue.start();
```

--------------------------------

### Prisma Client Initialization with @prisma/adapter-pg

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

Example of initializing a Prisma client instance using the @prisma/adapter-pg driver adapter, which is required for PrismaQueue.

```typescript
import { PrismaClient } from "@prisma/client";
import { PrismaPg } from "@prisma/adapter-pg";

const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL });
export const prisma = new PrismaClient({ adapter });
```

--------------------------------

### Install Prisma dependencies for v2.0

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

To use Prisma Queue v2.0, ensure you have the correct peer dependencies installed, including `@prisma/client` and `@prisma/adapter-pg` version 7 or higher.

```bash
pnpm add @prisma/client @prisma/adapter-pg pg
```

--------------------------------

### start()

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Starts the queue's polling and processing loop. Jobs will begin being dequeued and processed according to the configured concurrency and intervals. Safe to call multiple times.

```APIDOC
## start()

### Description
Starts the queue's polling and processing loop. Jobs will begin being dequeued and processed according to the configured concurrency and intervals.

### Method
`async start(): Promise<void>`

### Parameters
None

### Return Type
Promise that resolves when the polling loop has started.

### Throws
No exceptions; safe to call multiple times (subsequent calls are ignored if already running).

### Example
```typescript
const queue = createQueue({ prisma, name: "email" }, worker);
await queue.start();
console.log("Queue is now processing jobs");
```
```

--------------------------------

### Example RetryStrategy Implementation

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/types.md

Provides an example of a custom RetryStrategy function that implements exponential backoff, returning the delay in milliseconds or null to stop retrying.

```typescript
const strategy: RetryStrategy = ({ attempts, maxAttempts, error }) => {
  // Exponential backoff up to 5 attempts
  if (maxAttempts && attempts >= maxAttempts) return null;
  return 1000 * Math.pow(2, attempts);
};
```

--------------------------------

### Enqueue and Process Jobs

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Demonstrates how to enqueue a new job and start the queue processor. Includes a SIGTERM handler for graceful shutdown.

```typescript
// Enqueue a job
const job = await emailQueue.enqueue({ email: "user@example.com" });

// Start processing
await emailQueue.start();

// In your shutdown handler
process.on("SIGTERM", async () => {
  await emailQueue.stop({ timeout: 30_000 });
  await prisma.$disconnect();
});
```

--------------------------------

### Example: Updating Job Progress During Processing

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Demonstrates how to update a job's progress incrementally within a processing loop. This example iterates through items, processes each, and updates the job's progress percentage accordingly.

```typescript
const queue = createQueue(
  { prisma, name: "import" },
  async (job, client) => {
    const items = await fetchItems(job.payload.source);
    for (let i = 0; i < items.length; i++) {
      await processItem(items[i]);
      const percent = Math.round(((i + 1) / items.length) * 100);
      await job.progress(percent);
    }
    return { processed: items.length };
  }
);
```

--------------------------------

### Example: Manually Completing a Job

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Demonstrates how to manually update a job's status to complete by setting `finishedAt`, `progress`, and `result` fields.

```typescript
// Manually complete a job
await job.update({
  finishedAt: new Date(),
  progress: 100,
  result: { custom: "result" }
});
```

--------------------------------

### Example: Manually Removing a Job

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Shows how to explicitly delete a job from the database using the `delete()` method and log its ID.

```typescript
// Manually remove a job
const deleted = await job.delete();
console.log(`Deleted job ${deleted.id}`);
```

--------------------------------

### Start Queue Processing

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Initiate the processing of jobs in the queue. This is typically run in a separate process to handle background tasks.

```typescript
import { emailQueue } from "./emailQueue";

const main = async () => {
  await queue.start();
};

main();

```

--------------------------------

### Example: Handling Deleted Jobs After Fetching

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Shows how to check if a job still exists in the database after calling `fetch()`. If `fetch()` returns null, it indicates the job was deleted.

```typescript
const refreshed = await job.fetch();
if (refreshed === null) {
  console.log("Job was deleted from the database");
} else {
  console.log(`Job attempts: ${refreshed.attempts}`);
}
```

--------------------------------

### Enqueue Job with JobCreator Example

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/types.md

Demonstrates how to enqueue a job using the JobCreator type. The payload is computed asynchronously within a transaction using a provided transaction client.

```typescript
const job = await queue.enqueue(
  async (txClient) => {
    // Run in a transaction
    const user = await txClient.user.findUniqueOrThrow({ where: { id: 1 } });
    return { email: user.email, name: user.name };
  }
);
```

--------------------------------

### Schedule Regular Syncs with Interval

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Set up recurring jobs at a fixed interval, like hourly syncs. Use `repeatFrom: "finishedAt"` to tolerate slow jobs and ensure the next run starts after the previous one completes.

```typescript
await queue.schedule(
  {
    key: "hourly-sync",
    interval: { hours: 1 },
    repeatFrom: "finishedAt"  // Tolerates slow jobs
  },
  { source: "external-api" }
);
```

--------------------------------

### Example IntervalDuration Usage

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/types.md

Illustrates how to construct an IntervalDuration object to represent a specific time span, such as 1 day, 2 hours, and 30 minutes.

```typescript
// 1 day + 2 hours + 30 minutes
const duration: IntervalDuration = {
  days: 1,
  hours: 2,
  minutes: 30,
};
```

--------------------------------

### Example: Checking for Concurrent Job Processing

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Illustrates how to check if a job is currently locked by another worker. This is particularly relevant in transactional mode to prevent race conditions.

```typescript
// Check if another worker is processing this job
const locked = await job.isLocked();
if (locked) {
  console.log("Job is being processed by another worker");
}
```

--------------------------------

### Start PrismaQueue Processing

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Call this method to begin the queue's polling and job processing loop. Jobs will be dequeued and processed based on the configured concurrency and intervals. It's safe to call multiple times.

```typescript
const queue = createQueue({ prisma, name: "email" }, worker);
await queue.start();
console.log("Queue is now processing jobs");
```

--------------------------------

### Custom Retry Strategy with Linear Backoff

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Implement a custom retry strategy for failed jobs. This example uses linear backoff, retrying jobs at 1-second intervals for each attempt, up to the maximum attempts.

```ts
import { createQueue, type RetryContext } from "@mgcrea/prisma-queue";

const queue = createQueue<JobPayload, JobResult, typeof prisma>(
  {
    prisma,
    name: "email",
    maxAttempts: 5,
    retryStrategy: ({ attempts, maxAttempts, error }: RetryContext) => {
      // Return delay in ms, or null to stop retrying
      if (maxAttempts !== null && attempts >= maxAttempts) return null;
      // Linear backoff: 1s, 2s, 3s, ...
      return 1000 * attempts;
    },
  },
  async (job, client) => {
    // ...
  },
);
```

--------------------------------

### Configure Per-Job Timeout (5 Minutes)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

Set a `jobTimeout` to enforce a maximum execution time for individual jobs in non-transactional mode. This example sets a 5-minute limit, aborting the job's signal upon timeout. Workers should listen to the `abort` signal for cooperative cancellation.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "imports",
    jobTimeout: 5 * 60 * 1000,  // Abort after 5 minutes
  },
  async (job, client) => {
    job.signal.addEventListener("abort", () => {
      console.log("Job timeout reached, stopping work");
      // Clean up resources
    });

    try {
      await doWork({ signal: job.signal });
    } catch (error) {
      if (error.message.includes("aborted")) {
        console.log("Job was cancelled due to timeout");
      }
      throw error;
    }
  }
);
```

--------------------------------

### Schedule a fixed wall-clock interval job

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Schedules a job to run at fixed intervals based on the wall clock, preventing drift. This example checks status every 30 minutes.

```typescript
await queue.schedule(
  {
    key: "heartbeat",
    interval: { minutes: 30 },
    repeatFrom: "runAt"
  },
  { nodeId: "server-1" }
);
```

--------------------------------

### Configure Exactly-Once, Short Atomic Operations

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

This setup ensures that short operations are executed exactly once, even in the face of retries. It enables transactional behavior and sets a timeout for the transaction itself. A higher number of maximum attempts is used to increase the likelihood of successful atomic execution.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "atomic-updates",
    transactional: true,
    transactionTimeout: 30_000,
    maxAttempts: 5,
  },
  worker
);
```

--------------------------------

### Schedule a recurring job with interval

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Schedules a recurring job using a fixed interval. This example syncs every 20 hours, tolerating long-running jobs by rescheduling after completion.

```typescript
await queue.schedule(
  { key: "hourly-sync", interval: { hours: 20 } },
  { source: "upstream" }
);
```

--------------------------------

### Get queue statistics

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Retrieves a snapshot of job counts by state (pending, scheduled, processing, completed, dead). Useful for monitoring and alerts.

```typescript
const { pending, scheduled, processing, completed, dead } = await queue.stats();
console.log(`Queue status:
  Pending: ${pending}
  Scheduled: ${scheduled}
  Processing: ${processing}
  Completed: ${completed}
  Dead-lettered: ${dead}`);

// Alert if dead-letter depth is growing
if (dead > 100) {
  console.warn("High dead-letter count detected");
}
```

--------------------------------

### Configure Shorter Stale Job Reclamation (5 Minutes)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

Use a shorter `staleTimeout` for jobs that are expected to complete quickly. This example sets the reclamation period to 5 minutes, suitable for fast-running tasks.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "api-calls",
    staleTimeout: 5 * 60 * 1000,  // Reclaim after 5 minutes stuck
  },
  worker
);
```

--------------------------------

### Schedule an Interval Job (Drift-Tolerant)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Schedules a job to run at a fixed interval after the previous job finishes. This mode is drift-tolerant, meaning the interval starts from the completion time.

```typescript
// Drift-tolerant: runs 20 hours after the job finishes
await queue.schedule(
  { key: "sync", interval: { hours: 20 } },
  { source: "upstream" }
);
```

--------------------------------

### Generate Prisma Client

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Run the Prisma generate command to create your Prisma Client instance.

```bash
prisma generate
```

--------------------------------

### Configuring Separate Queues by Name

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

Demonstrates how to create distinct queues for different job types by specifying a unique 'name' option for each.

```typescript
// Separate queues for different job types
const emailQueue = createQueue({ prisma, name: "email" }, emailWorker);
const reportQueue = createQueue({ prisma, name: "reports" }, reportWorker);
```

--------------------------------

### Get queue size

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Returns the total number of unfinished jobs. Optionally, filter to count only immediately runnable jobs.

```typescript
// Total unfinished jobs
const total = await queue.size();

// Jobs ready to run right now
const ready = await queue.size(true);

console.log(`${ready} of ${total} jobs are ready`);
```

--------------------------------

### Apply Prisma Schema Changes

Source: https://github.com/mgcrea/prisma-queue/blob/master/MIGRATION.md

Generate the Prisma client and push the schema changes to the database, or use `prisma migrate dev` for a managed migration.

```bash
prisma generate
prisma db push   # or: prisma migrate dev --name prisma_queue_v3
```

--------------------------------

### Create Prisma Queue with Event Listeners

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-create-queue.md

Sets up a queue with event listeners for monitoring and observability. Listeners can be attached for various lifecycle events like enqueue, dequeue, success, errors, and more.

```typescript
const queue = createQueue(
  { prisma, name: "events" },
  async (job, client) => {
    await processJob(job.payload);
    return { done: true };
  }
);

queue.on("enqueue", (job) => {
  console.log(`Job ${job.id} enqueued`);
});

queue.on("dequeue", (job) => {
  console.log(`Job ${job.id} started processing`);
});

queue.on("success", (result, job) => {
  console.log(`Job ${job.id} succeeded:`, result);
});

queue.on("jobError", (error, job) => {
  console.error(`Job ${job.id} failed:`, error);
});

queue.on("dead", (error, job) => {
  console.error(`Job ${job.id} dead-lettered:`, error);
});

queue.on("reclaim", (jobs) => {
  console.log(`Reclaimed ${jobs.length} stale jobs`);
});

queue.on("error", (error) => {
  console.error("Queue system error:", error);
});

```

--------------------------------

### Accessing Complete Job Database Record

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Get the entire job record as stored in the database. This includes all fields, such as attempts, maxAttempts, progress, and more.

```typescript
const {
  attempts,
  maxAttempts,
  progress,
  notBefore,
  processedAt,
  ...other
} = job.record;
console.log(`Attempts: ${attempts}/${maxAttempts}, Progress: ${progress}%`);
```

--------------------------------

### Accessing Interval Schedule in Milliseconds

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Get the interval in milliseconds for interval-scheduled jobs. Convert to seconds or other units as needed. Returns null if not interval-scheduled.

```typescript
if (job.intervalMs) {
  const intervalSecs = Number(job.intervalMs) / 1000;
  console.log(`Runs every ${intervalSecs} seconds`);
}
```

--------------------------------

### Instantiate queue with PrismaClient in v2.0

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

In v2.0, the `prisma` option is required when creating a queue instance. You must pass your configured `PrismaClient` to the `createQueue` function.

```typescript
// v1
// const queue = createQueue({ name: "email" }, worker);

// v2
import { prisma } from "./prisma";
const queue = createQueue({ prisma, name: "email" }, worker);
```

--------------------------------

### Get Queue Statistics

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Retrieves the current counts of jobs in different states within the queue. Useful for monitoring queue health and depth.

```javascript
const { pending, scheduled, processing, completed, dead } = await queue.stats();
console.log(`Dead-letter depth: ${dead}`);
```

--------------------------------

### Create a TypeScript Queue Instance

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Define a new queue with its configuration and a job processing function. Ensure your PrismaClient instance is imported.

```typescript
import { createQueue } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma";  // your PrismaClient instance

type EmailPayload = { email: string };
type EmailResult = { messageId: string };

export const emailQueue = createQueue<EmailPayload, EmailResult, typeof prisma>(
  {
    prisma,
    name: "email",
    maxConcurrency: 5,
    maxAttempts: 3,
  },
  async (job, client) => {
    console.log(`Sending email to ${job.payload.email}`);
    const { messageId } = await sendEmail(job.payload.email);
    return { messageId };
  }
);
```

--------------------------------

### Prisma Queue Project File Structure

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

This snippet details the directory and file organization within the Prisma Queue project. It helps understand where different modules and utilities are located.

```tree
src/
├── index.ts                    # Main exports, createQueue factory
├── PrismaQueue.ts             # Main queue class
├── PrismaJob.ts               # Job instance class
├── types.ts                   # TypeScript type definitions
├── errors.ts                  # JobExhaustedError
└── utils/
    ├── index.ts               # Utility exports
    ├── time.ts                # waitFor, calculateDelay, intervalToMs
    ├── error.ts               # serializeError, isPrismaError
    ├── string.ts              # escape, capitalize, uncapitalize
    ├── stringify.ts           # prepareForJson, restoreFromJson
    └── debug.ts               # debug logger
```

--------------------------------

### Accessing Job Priority

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Get the priority level of a job. Lower values indicate higher priority. Ties are broken by the job's runAt time.

```typescript
queue.on("dequeue", (job) => {
  console.log(`Processing job with priority ${job.priority}`);
});
```

--------------------------------

### Job Methods (via PrismaJob)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Methods available on individual job objects obtained from the queue.

```APIDOC
## Job Methods (via PrismaJob)

### `progress(value)`

#### Description
Updates the progress percentage of the current job.

### `fetch()`

#### Description
Refreshes the job's data from the database, ensuring you have the latest information.

### `update(data)`

#### Description
Updates specific fields of the job record in the database.

### `delete()`

#### Description
Removes the job from the database.

### `isLocked()`

#### Description
Checks if the job is currently locked by an active processing transaction.
```

--------------------------------

### Create a Queue Instance

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Define a new queue with job payload and result types, and a worker function to process jobs. The worker receives the job details and a Prisma client instance.

```typescript
import { createQueue } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma"; // your PrismaClient instance

type JobPayload = { email: string };
type JobResult = { status: number };

export const emailQueue = createQueue<JobPayload, JobResult, typeof prisma>(
  { prisma, name: "email" },
  async (job, client) => {
    const { id, payload } = job;
    console.log(`Processing job#${id} with payload=${JSON.stringify(payload)})`);
    // await someAsyncMethod();
    await job.progress(50);
    const status = 200;
    if (Math.random() > 0.5) {
      throw new Error(`Failed for some unknown reason`);
    }
    console.log(`Finished job#${id} with status=${status}`);
    return { status };
  },
);

```

--------------------------------

### Prisma Queue Event Listeners for Observability

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Set up event listeners for various queue events to integrate with metrics, tracing, and alerting systems. This provides comprehensive observability into queue operations.

```typescript
queue.on("enqueue", (job) => {
  metrics.increment("queue.enqueued", { queue: job.queue });
  tracing.log(`Enqueued job ${job.id}`);
});

queue.on("dequeue", (job) => {
  metrics.gauge("queue.processing", 1, { queue: job.queue });
  tracing.startSpan(`job_${job.id}`);
});

queue.on("success", (result, job) => {
  metrics.increment("queue.success", { queue: job.queue });
  metrics.timing("job.duration", Date.now() - job.createdAt.getTime());
  tracing.endSpan(`job_${job.id}`, { success: true });
});

queue.on("jobError", (error, job) => {
  metrics.increment("queue.attempt_failed", { queue: job.queue });
  tracing.log(`Job ${job.id} attempt failed: ${error.message}`);
});

queue.on("dead", (error, job) => {
  metrics.increment("queue.dead_lettered", { queue: job.queue });
  metrics.gauge("queue.dlq_depth", 1, { queue: job.queue }, "increment");
  alert.error(`Job ${job.id} permanently failed`, { job: job.id, error });
});

queue.on("reclaim", (jobs) => {
  metrics.increment("queue.reclaimed", jobs.length, { queue: queue.name });
  tracing.log(`Reclaimed ${jobs.length} stale jobs`);
});

queue.on("error", (error) => {
  metrics.increment("queue.system_error", { queue: queue.name });
  alert.critical("Queue system failure", error);
});
```

--------------------------------

### Create Basic Prisma Queue (At-Least-Once)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-create-queue.md

Configures a queue with default at-least-once semantics. The worker receives the full PrismaClient and runs outside the dequeue transaction. Suitable for general-purpose job processing.

```typescript
import { createQueue } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma";

type EmailPayload = { email: string; subject: string };
type EmailResult = { messageId: string };

const emailQueue = createQueue<EmailPayload, EmailResult, typeof prisma>(
  {
    prisma,
    name: "email",
    maxConcurrency: 5,
    pollInterval: 5000,
  },
  async (job, client) => {
    // client is the full PrismaClient — $transaction is available
    const result = await client.$transaction(async (tx) => {
      await tx.user.update({
        where: { email: job.payload.email },
        data: { lastEmailSent: new Date() },
      });
      const { messageId } = await sendEmail(job.payload);
      return { messageId };
    });
    return result;
  }
);

```

--------------------------------

### add

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Alias for `enqueue()`. Adds a job to the queue with identical parameters and behavior.

```APIDOC
## add(payloadOrFunction, options?)

### Description
Alias for `enqueue()`. Adds a job to the queue. Parameters and behavior are identical to `enqueue()`.

### Method
`POST` (inferred)

### Endpoint
`/queue/add` (inferred)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **payloadOrFunction** (`T | JobCreator<T, C>`) - Required - Job payload, or async function that returns payload (function runs in a transaction)
- **options** (`EnqueueOptions`) - Optional - Scheduling, priority, and attempt configuration (same as `enqueue`)

### Request Example
```json
{
  "payloadOrFunction": {"email": "user@example.com"},
  "options": {"priority": 1}
}
```

### Response
#### Success Response (200)
- **PrismaJob<T, U>** - The newly created job.

#### Response Example
```json
{
  "id": "job_id",
  "payload": {},
  "status": "pending",
  "createdAt": "<timestamp>",
  "runAt": "<timestamp>"
}
```
```

--------------------------------

### Implement Custom Job Retry Strategy

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Define a custom retry strategy to control how failed jobs are retried. This example skips retries for client errors (4xx) and applies exponential backoff with jitter for other errors.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "api-calls",
    retryStrategy: ({ attempts, maxAttempts, error }) => {
      // Don't retry client errors (4xx)
      if (error instanceof Error && /^4\d{2}/.test(error.message)) {
        return null;
      }
      // Retry with jitter
      if (maxAttempts && attempts >= maxAttempts) return null;
      return Math.random() * 1000 * Math.pow(2, attempts);
    },
  },
  worker
);
```

--------------------------------

### Manually Add `deadLetteredAt` and Partial Index (SQL)

Source: https://github.com/mgcrea/prisma-queue/blob/master/MIGRATION.md

If the `partialIndexes` preview feature is not enabled, manually add the `deadLetteredAt` column and create a partial index using raw SQL. This includes dropping the old full index if it exists.

```sql
-- DateTime maps to timestamp(3) (without time zone) by default, matching the other columns
ALTER TABLE queue_jobs ADD COLUMN "deadLetteredAt" timestamp(3);

-- Drop the v2 full index, then create the partial one. Confirm the old name first with `\d queue_jobs`
-- (Prisma's default is queue_jobs_queue_finishedAt_processedAt_priority_runAt_idx).
DROP INDEX IF EXISTS queue_jobs_queue_finishedAt_processedAt_priority_runAt_idx;
CREATE INDEX queue_jobs_queue_processedAt_priority_runAt_idx
  ON queue_jobs (queue, "processedAt", priority, "runAt")
  WHERE "finishedAt" IS NULL;
```

--------------------------------

### Handle Worker Crashes and Job Reclaim

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/errors.md

This example shows how to handle worker crashes in non-transactional mode using `staleTimeout`. If a worker crashes, the job claim becomes stale, and after the timeout, the job can be reclaimed and retried. The 'dead' event is also monitored for `JobExhaustedError`.

```typescript
const queue = createQueue(
  {
    prisma,
    name: "work",
    maxAttempts: 2,
    staleTimeout: 30_000,  // 30 seconds
  },
  async (job, client) => {
    if (Math.random() > 0.5) {
      process.exit(1);  // Simulate crash
    }
    return { done: true };
  }
);

queue.on("reclaim", (jobs) => {
  console.log(`Reclaimed ${jobs.length} stuck jobs`);
});

queue.on("dead", (error, job) => {
  if (error instanceof JobExhaustedError) {
    console.error(`Job ${job.id} was orphaned and exhausted`);
  }
});

await queue.start();
```

--------------------------------

### Add QueueJob Model to Prisma Schema

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Append the `QueueJob` model to your `prisma/schema.prisma` file to manage queue jobs.

```prisma
generator client {
  provider        = "prisma-client"
  previewFeatures = ["partialIndexes"]
}

datasource db {
  provider = "postgresql"
}

model QueueJob {
  id          BigInt    @id @default(autoincrement())
  queue       String
  key         String?
  cron        String?
  intervalMs  BigInt?
  repeatFrom  String?
  payload     Json?
  result      Json?
  error       Json?
  progress    Int       @default(0)
  priority    Int       @default(0)
  attempts    Int       @default(0)
  maxAttempts Int?
  runAt       DateTime  @default(now())
  notBefore   DateTime?
  finishedAt  DateTime?
  processedAt DateTime?
  failedAt    DateTime?
  deadLetteredAt DateTime?
  createdAt   DateTime  @default(now())
  updatedAt   DateTime  @updatedAt

  @@unique([queue, key, runAt])
  @@index([queue, processedAt, priority, runAt], where: raw("""finishedAt" IS NULL"""))
  @@map("queue_jobs")
}
```

--------------------------------

### Handling Job Cancellation with AbortSignal

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Demonstrates how to use the AbortSignal provided by PrismaJob for cooperative cancellation. It includes setting up an event listener for 'abort' and passing the signal to an asynchronous operation.

```typescript
const queue = createQueue(
  { prisma, name: "imports", jobTimeout: 5 * 60 * 1000 },
  async (job, client) => {
    job.signal.addEventListener("abort", () => {
      console.log("Job was cancelled, cleaning up...");
    });

    try {
      await importData(job.payload, { signal: job.signal });
    } catch (error) {
      if (error.name === "AbortError") {
        console.log("Import was cancelled");
      }
      throw error;
    }
  }
);
```

--------------------------------

### Schedule Recurring Jobs with Interval (Drift-Free)

Source: https://github.com/mgcrea/prisma-queue/blob/master/README.md

Schedule jobs with a fixed wall-clock cadence using `repeatFrom: 'runAt'`. This ensures jobs run at consistent intervals regardless of job duration.

```typescript
// Drift-free: every run lands on a 20-hour boundary from the first runAt
await queue.schedule(
  { key: "sync", interval: { hours: 20 }, repeatFrom: "runAt" },
  { source: "upstream" },
);
```

--------------------------------

### job.repeatFrom

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Retrieves the anchor for interval-based rescheduling. It can be 'finishedAt' (default, drift-tolerant) or 'runAt' (fixed wall-clock cadence).

```APIDOC
## job.repeatFrom

### Description

The anchor for interval-based rescheduling: "finishedAt" (default, drift-tolerant) or "runAt" (fixed wall-clock cadence).

### Method

GET

### Endpoint

N/A (Getter method on Job object)

### Parameters

None

### Response

#### Success Response

- **repeatFrom** (string | null) - The rescheduling anchor ('finishedAt' or 'runAt') or null.

### Response Example

```json
"finishedAt"
```
```

--------------------------------

### Create Prisma Queue with Per-Job Timeout

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-create-queue.md

Configures a queue with a per-job timeout, useful for aborting long-running jobs in non-transactional mode. The job's signal can be used to cancel ongoing operations.

```typescript
const importQueue = createQueue(
  {
    prisma,
    name: "imports",
    jobTimeout: 5 * 60 * 1000, // 5 minutes
  },
  async (job, client) => {
    // job.signal aborts on timeout or queue.stop()
    job.signal.addEventListener("abort", () => {
      console.log("Job was cancelled");
    });

    await importData(job.payload, { signal: job.signal });
    return { imported: true };
  }
);

```

--------------------------------

### Queue Lifecycle Management

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Methods for controlling the overall lifecycle of the queue processor.

```APIDOC
## Queue Lifecycle

### `start()`

#### Description
Starts the queue processor, enabling it to poll for and process jobs.

### `stop(options?)`

#### Description
Stops the queue processor gracefully. It waits for any currently executing jobs to complete before shutting down.
```

--------------------------------

### Implement Queue Event Observability

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Monitor queue activity by subscribing to events like 'enqueue', 'success', 'dead', and 'error'. Use this to track metrics, log errors, and trigger alerts.

```typescript
queue.on("enqueue", (job) => metrics.increment("queue.enqueued"));
queue.on("success", (result, job) => {
  metrics.increment("queue.success");
  metrics.timing("job.duration", Date.now() - job.createdAt.getTime());
});
queue.on("dead", (error, job) => {
  if (error instanceof JobExhaustedError) {
    alert.error(`Dead-letter: ${job.queue}/${job.id}`);
  }
});
queue.on("error", (error) => alert.critical("Queue failure", error));
```

--------------------------------

### Create Transactional Prisma Queue (Exactly-Once)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-create-queue.md

Configures a queue for exactly-once semantics. The worker runs within the dequeue transaction using a transaction-scoped client. Ideal for short, atomic operations that must succeed or fail with the dequeue transaction.

```typescript
import { createQueue, type JobWorker } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma";

type ProcessingPayload = { dataId: number };
type ProcessingResult = { processed: boolean };

const processingQueue = createQueue<ProcessingPayload, ProcessingResult, typeof prisma>(
  {
    prisma,
    name: "processing",
    transactional: true,
    transactionTimeout: 60_000,
  },
  async (job, txClient) => {
    // txClient is transaction-scoped, no $transaction available
    const data = await txClient.data.findUniqueOrThrow({
      where: { id: job.payload.dataId },
    });
    const result = await processData(data);
    await txClient.data.update({
      where: { id: job.payload.dataId },
      data: { processed: true, result: JSON.stringify(result) },
    });
    return { processed: true };
  }
);

```

--------------------------------

### fetch()

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Retrieves the latest job record from the database and updates the internal state. Returns null if the job record no longer exists.

```APIDOC
## fetch()

### Description
Retrieves the latest job record from the database and updates the internal state. Returns null if the row no longer exists (e.g., deleted by `deleteOn` configuration).

### Method
`fetch`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Return Type
Promise<DatabaseJob<Payload, Result> | null> - Promise resolving to the job record, or null if deleted.

### Example
```typescript
const refreshed = await job.fetch();
if (refreshed === null) {
  console.log("Job was deleted from the database");
} else {
  console.log(`Job attempts: ${refreshed.attempts}`);
}
```
```

--------------------------------

### Create Prisma Queue with Custom Retry Strategy

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-create-queue.md

Configures a queue with a custom retry strategy, overriding the default exponential backoff. This allows for specific retry intervals based on attempts and errors.

```typescript
import { createQueue, type RetryContext } from "@mgcrea/prisma-queue";
import { prisma } from "./prisma";

const customQueue = createQueue(
  {
    prisma,
    name: "custom-retry",
    maxAttempts: 3,
    retryStrategy: ({ attempts, maxAttempts, error }: RetryContext) => {
      // Linear backoff: 1s, 2s, 3s
      if (maxAttempts && attempts >= maxAttempts) return null;
      return 1000 * attempts;
    },
  },
  async (job, client) => {
    await doWork();
    return { success: true };
  }
);

```

--------------------------------

### Enqueue Job with Unique Key and RunAt

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/errors.md

Illustrates attempting to enqueue a job with a specific key and runAt timestamp. The second enqueue call with the same key and runAt will fail due to a unique constraint violation.

```typescript
// First call succeeds
await queue.enqueue(
  { email: "user@example.com" },
  { key: "send-email", runAt: new Date("2024-01-15") }
);

// Second call with same key and runAt fails
await queue.enqueue(
  { email: "user@example.com" },
  { key: "send-email", runAt: new Date("2024-01-15") }
);
// Error: Unique constraint failed on (queue, key, runAt)
```

--------------------------------

### calculateDelay(attempts, maxDelay?)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/utilities.md

Computes exponential-backoff delay with jitter for a given attempt count. This is useful for implementing retry logic with increasing delays.

```APIDOC
## calculateDelay(attempts, maxDelay?)

### Description
Computes exponential-backoff delay with jitter for a given attempt count.

### Parameters
#### Path Parameters
- **attempts** (number) - Required - Number of attempts already made (1-indexed)
- **maxDelay** (number) - Optional - Default: MAX_TIMEOUT_DELAY - Ceiling for the delay

### Return Type
`number` — Delay in milliseconds

### Formula
`min(2^attempts * 1000 + jitter(0–100ms), maxDelay)`

### Example
```typescript
import { calculateDelay } from "@mgcrea/prisma-queue/utils";

calculateDelay(1);  // ~2000–2100ms (2^1 * 1000)
calculateDelay(2);  // ~4000–4100ms (2^2 * 1000)
calculateDelay(3);  // ~8000–8100ms (2^3 * 1000)
calculateDelay(5, 30_000);  // ~30000ms (capped at maxDelay)
```
```

--------------------------------

### Enqueue a Scheduled Job

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Schedule a job to run at a specific time by providing a `runAt` option. The job will be added to the queue but will not be processed until the specified date and time.

```typescript
const scheduledJob = await queue.enqueue(
  { email: "admin@example.com" },
  { runAt: new Date(Date.now() + 60_000) } // Run in 1 minute
);

```

--------------------------------

### Job Operations

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

Methods for adding and managing jobs within the queue.

```APIDOC
## Job Operations

### `enqueue(payload, options?)`

#### Description
Adds a single job to the queue with the specified payload and optional configuration.

### `add(payload, options?)`

#### Description
An alias for the `enqueue()` method, used to add a single job to the queue.

### `enqueueMany(payloads, options?)`

#### Description
Adds multiple jobs to the queue in a bulk operation, optimized for high throughput.

### `schedule(options, payload)`

#### Description
Schedules a recurring job to be executed at a defined interval or cron schedule.
```

--------------------------------

### stats()

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Returns a snapshot of job counts by state: pending, scheduled, processing, completed, and dead. Useful for dashboards and alerts.

```APIDOC
## stats()

### Description
Returns a snapshot of job counts by state: pending (ready to run), scheduled (waiting), processing (claimed), completed (succeeded), and dead (dead-lettered). Useful for dashboards and alerts.

### Method
GET

### Endpoint
/queue/stats

### Parameters
None

### Request Example
None

### Response
#### Success Response (200)
- **pending** (number) - Count of jobs ready to run now.
- **scheduled** (number) - Count of jobs waiting for their scheduled run time.
- **processing** (number) - Count of jobs currently being processed.
- **completed** (number) - Count of jobs that finished successfully.
- **dead** (number) - Count of jobs that failed permanently.

#### Response Example
```json
{
  "pending": 10,
  "scheduled": 5,
  "processing": 2,
  "completed": 100,
  "dead": 1
}
```
```

--------------------------------

### Prisma Partial Index for Active Jobs

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/README.md

This Prisma schema definition creates a partial index to optimize dequeue queries by only including unfinished jobs. It requires the `partialIndexes` preview feature.

```prisma
@@index([queue, processedAt, priority, runAt], where: raw ("finishedAt" IS NULL))
```

--------------------------------

### PrismaQueue Options Type Definition

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/configuration.md

Defines the structure for the PrismaQueue configuration object, outlining all available properties and their types.

```typescript
export type PrismaQueueOptions<C = unknown> = {
  prisma: C;
  name?: string;
  maxAttempts?: number | null;
  maxConcurrency?: number;
  pollInterval?: number;
  jobInterval?: number;
  tableName?: string;
  deleteOn?: "success" | "always" | "never";
  transactionTimeout?: number;
  transactionWarningTimeout?: number | null;
  retryStrategy?: RetryStrategy;
  maxRetryDelay?: number;
  transactional?: boolean;
  staleTimeout?: number | null;
  jobTimeout?: number | null;
};
```

--------------------------------

### PrismaJobOptions Type Definition

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-job.md

Defines the configuration options for Prisma jobs, including the Prisma model delegate, client, table name, and an optional abort signal.

```typescript
export type PrismaJobOptions = {
  model: any;          // Prisma queueJob delegate
  client: any;         // Prisma client or transaction-scoped client
  tableName: string;   // Escaped table name for raw SQL
  signal?: AbortSignal; // Optional abort signal for cancellation
};
```

--------------------------------

### Update Prisma Schema for v3.0

Source: https://github.com/mgcrea/prisma-queue/blob/master/MIGRATION.md

Add the `deadLetteredAt` column and a partial index to the `QueueJob` model. Ensure the `partialIndexes` preview feature is enabled for Prisma versions 7.4+.

```prisma
generator client {
  provider        = "prisma-client"
  output          = "./client"
  previewFeatures = ["partialIndexes"] // required for the partial @@index below (Prisma 7.4+)
}

model QueueJob {
  // ...existing fields...
  deadLetteredAt DateTime?

  @@unique([queue, key, runAt])
  // Replaces the v2 full index. Only indexes pending rows.
  @@index([queue, processedAt, priority, runAt], where: raw("\"finishedAt\" IS NULL"))
  @@map("queue_jobs")
}
```

--------------------------------

### schedule(options, payloadOrFunction)

Source: https://github.com/mgcrea/prisma-queue/blob/master/_autodocs/api-reference-prisma-queue.md

Schedules a recurring job using either a cron expression or a fixed interval. The job runs according to the schedule and is automatically rescheduled after each completion.

```APIDOC
## schedule(options, payloadOrFunction)

### Description
Schedules a recurring job using either a cron expression or a fixed interval. The job runs according to the schedule and is automatically rescheduled after each completion.

### Method
POST

### Endpoint
/queue/schedule

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **options** (ScheduleOptions) - Required - Cron or interval configuration (one, not both)
- **payloadOrFunction** (T | JobCreator<T, C>) - Required - Job payload or function that returns it

### Request Example
```json
{
  "options": {
    "key": "daily-report",
    "cron": "0 6 * * *"
  },
  "payloadOrFunction": {
    "recipientEmail": "ops@example.com"
  }
}
```

### Response
#### Success Response (200)
- **job** (PrismaJob<T, U>) - The scheduled job.

#### Response Example
```json
{
  "job": {
    "id": "job_id_123",
    "key": "daily-report",
    "status": "scheduled",
    "runAt": "2023-10-27T06:00:00.000Z",
    "createdAt": "2023-10-26T10:00:00.000Z"
  }
}
```

### ScheduleOptions (Cron Variant)
```json
{
  "key": "string",
  "cron": "string",
  "runAt": "Date",
  "maxAttempts": "number",
  "priority": "number"
}
```

### ScheduleOptions (Interval Variant)
```json
{
  "key": "string",
  "interval": {
    "days": "number",
    "hours": "number",
    "minutes": "number",
    "seconds": "number"
  },
  "repeatFrom": "finishedAt" | "runAt",
  "runAt": "Date",
  "maxAttempts": "number",
  "priority": "number"
}
```

### Throws
- If both `cron` and `interval` are provided
- If neither `cron` nor `interval` is provided
- If the cron expression is invalid
```