# OpenSpec Change: enterprise-queue

- **Change ID**: enterprise-queue
- **Package**: `@suaveplan/queue`
- **Category**: universal
- **Target path**: `universal/streams/queue`
- **Status**: PROPOSED
- **Date**: 2026-04-24
- **Reference spec**: `/mnt/s/web/suaveplan/enterprise/queue.md`

---

## Summary

Deliver `@suaveplan/queue` on the genesis branch from scratch at enterprise quality. The package provides queue management primitives for task/job processing: a typed `Queue` interface, in-memory FIFO/priority implementations, `QueueWorker` for concurrent job processing with bounded concurrency, retry/dead-letter policies, `AbortSignal` cancellation, event emission for job lifecycle, OTel instrumentation, and Zod-validated configuration. This is the single-process queue substrate; `@suaveplan/queue-distribution` builds the distributed layer on top.

---

## Current-state audit (genesis)

The package does not exist on genesis yet — fresh build. No prior source, no test results, no coverage data.

---

## Available workspace deps on genesis

| Package | Role |
|---|---|
| `@suaveplan/error` (`workspace:^`) | Base class for `QueueError`; `defineCodes` for `QUEUE.*` codes |
| `@suaveplan/types` (`workspace:^`) | `LoggerLike`, `Clock`, `EventBusLike`, `IdGenerator` injection interfaces |
| `@suaveplan/events` (`workspace:^`) | Job lifecycle event emission (`job:enqueued`, `job:started`, `job:completed`, `job:failed`, `job:dead`) |
| `zod` (`catalog:forms`) | Zod schemas for all public option and job types |
| `@opentelemetry/api` (`catalog:observability`) | Spans, counters, histograms on enqueue, process, and retry operations |
| `tslib` (`catalog:build`) | Runtime helpers |

Dev deps: `vitest`, `@vitest/coverage-istanbul`, `fast-check`, `mitata` (all `catalog:testing`).

---

## Public API surface

### Queue interface

```ts
interface Queue<T> {
  readonly name: string
  readonly size: number
  enqueue(job: Job<T>): Promise<string>          // returns jobId
  dequeue(): Promise<Job<T> | undefined>
  peek(): Promise<Job<T> | undefined>
  get(jobId: string): Promise<Job<T> | undefined>
  ack(jobId: string): Promise<void>
  nack(jobId: string, reason?: string): Promise<void>
  remove(jobId: string): Promise<boolean>
  clear(): Promise<void>
  drain(): Promise<void>
}

interface Job<T> {
  id: string
  payload: T
  priority?: number           // higher = processed first; default: 0
  scheduledAt?: Date
  maxRetries?: number
  attemptCount: number
  createdAt: Date
  state: JobState
}

type JobState = "pending" | "processing" | "completed" | "failed" | "dead"
```

### Queue implementations

```ts
// In-memory FIFO queue
function createMemoryQueue<T>(options?: MemoryQueueOptions<T>): Queue<T>

// In-memory priority queue (max-heap on job.priority)
function createPriorityQueue<T>(options?: PriorityQueueOptions<T>): Queue<T>

// Delayed job queue (jobs are not dequeued before scheduledAt)
function createDelayedQueue<T>(options?: DelayedQueueOptions<T>): Queue<T>

interface MemoryQueueOptions<T> {
  name?: string
  maxSize?: number
  clock?: Clock
  idGenerator?: IdGenerator
  bus?: EventBusLike
  logger?: LoggerLike
}

type PriorityQueueOptions<T> = MemoryQueueOptions<T>
type DelayedQueueOptions<T> = MemoryQueueOptions<T> & { pollIntervalMs?: number }
```

### QueueWorker

```ts
function createQueueWorker<T>(options: QueueWorkerOptions<T>): QueueWorker<T>

interface QueueWorker<T> {
  start(signal?: AbortSignal): Promise<void>
  stop(): Promise<void>
  readonly stats: WorkerStats
}

interface QueueWorkerOptions<T> {
  queue: Queue<T>
  handler: JobHandler<T>
  concurrency?: number            // default: 1
  retryPolicy?: RetryPolicy
  deadLetterQueue?: Queue<T>
  logger?: LoggerLike
  clock?: Clock
}

type JobHandler<T> = (job: Job<T>, signal: AbortSignal) => Promise<void>

interface RetryPolicy {
  maxRetries: number
  backoff: "constant" | "linear" | "exponential"
  initialDelayMs: number
  maxDelayMs?: number
}

interface WorkerStats {
  processed: number
  succeeded: number
  failed: number
  dead: number
  retried: number
  avgProcessingMs: number
}
```

### Job lifecycle events (when bus provided)

```ts
"queue:job:enqueued"    // { queueName, jobId, priority }
"queue:job:started"     // { queueName, jobId, attempt }
"queue:job:completed"   // { queueName, jobId, durationMs }
"queue:job:failed"      // { queueName, jobId, error, attempt, willRetry }
"queue:job:dead"        // { queueName, jobId, reason }
```

### Error handling

```ts
class QueueError extends SuaveplanError<QueueErrorCode>

type QueueErrorCode =
  | "QUEUE_FULL"
  | "QUEUE_JOB_NOT_FOUND"
  | "QUEUE_WORKER_ABORTED"
  | "QUEUE_HANDLER_FAILED"
  | "QUEUE_INVALID_STATE"
  | "QUEUE_DEAD_LETTER_FAILED"

function registerQueueErrorCodes(): void
```

### Zod schemas

```ts
const JobSchema: ZodSchema
const RetryPolicySchema: ZodSchema
const MemoryQueueOptionsSchema: ZodSchema
const QueueWorkerOptionsSchema: ZodSchema
```

---

## New additions for genesis

### 1. Delayed queue with injectable clock

`createDelayedQueue` uses the injectable `Clock` for `scheduledAt` comparisons, enabling deterministic tests without real timers.

### 2. Dead letter queue

When a job exceeds `maxRetries`, it is moved to the `deadLetterQueue` (if configured) with the failure reason and full retry history attached. The dead letter queue is itself a `Queue<T>`, allowing inspection and manual replay.

### 3. OTel instrumentation

- `queue.enqueue` — span `queue.enqueue`, attributes: `queue.name`, `queue.priority`
- `worker.process` — span `queue.worker.process`, attributes: `queue.name`, `queue.job_id`, `queue.attempt`
- Counters: `queue.jobs_enqueued_total`, `queue.jobs_completed_total`, `queue.jobs_failed_total`, `queue.jobs_dead_total`
- Histogram: `queue.processing_duration_ms` (attribute: `queue.name`)
- Gauge: `queue.depth` (attribute: `queue.name`)

### 4. Property-based tests (fast-check)

- FIFO invariant: dequeue order matches enqueue order for equal-priority jobs
- Priority invariant: highest-priority job is always dequeued first
- Retry count bound: job.attemptCount never exceeds maxRetries + 1 before moving to dead letter
- Worker concurrency: at most `concurrency` jobs are in `processing` state at any time

### 5. Benchmark suite (`src/__benchmarks__/`)

- Enqueue + dequeue 100k items (memory queue)
- Worker with concurrency=10 processing 1k instant jobs

---

## File layout

```text
universal/streams/queue/
  package.json
  tsconfig.json
  vitest.config.ts
  src/
    index.ts
    error-codes.ts
    constants/
      index.ts
    types/
      common.ts
      common.md
    queues/
      memory-queue/
        memory-queue.ts
        memory-queue.md
        memory-queue.types.ts
        memory-queue.test.ts
        index.ts
      priority-queue/
        priority-queue.ts
        priority-queue.md
        priority-queue.types.ts
        priority-queue.test.ts
        index.ts
      delayed-queue/
        delayed-queue.ts
        delayed-queue.md
        delayed-queue.types.ts
        delayed-queue.test.ts
        index.ts
    worker/
      queue-worker/
        queue-worker.ts
        queue-worker.md
        queue-worker.types.ts
        queue-worker.test.ts
        index.ts
    utils/
      retry-policy.ts
      retry-policy.md
      retry-policy.test.ts
    __benchmarks__/
      queue.bench.ts
  tests/
    integration.test.ts
    property.test.ts
```

---

## Impact

**Downstream packages**: `@suaveplan/queue-distribution` extends this package with distributed worker pools. Any background-job, batch-processing, or task-scheduling feature in the monorepo uses this as the foundation.

**Workspace dep prerequisites**: `@suaveplan/error`, `@suaveplan/types`, `@suaveplan/events` must be published to Verdaccio.

**Breaking changes**: none (new package).