# bunqueue

"When I report a bug, don't start by trying to fix it. Instead, start by writing a test that reproduces the bug. Then, have subagents try to fix the bug and prove it with a passing test."

**MANDATORY: After ANY code modification, ALWAYS run ALL THREE test suites before committing:**

```bash
bun test                                # Unit tests (~5000 tests)
bun scripts/tcp/run-all-tests.ts        # TCP integration tests (~50 suites)
bun scripts/embedded/run-all-tests.ts   # Embedded integration tests (~35 suites)
```

Never commit without all three passing. No exceptions.

**MANDATORY: After every commit, ALWAYS:**

1. Bump version in `package.json`
2. Update changelog in `docs/src/content/docs/changelog.md`
3. `git push origin main`
4. `bun publish` to publish new version to npm

No exceptions. Every commit = new version + changelog + npm publish.

High-performance job queue server for Bun. SQLite persistence, cron jobs, priorities, DLQ, S3 backups.

## Architecture Flow

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                              CLIENT                                          │
│  Queue.add() ─────┐                              ┌───── Worker.process()    │
│  Queue.addBulk() ─┤                              │                          │
│                   ▼                              ▼                          │
│            ┌──────────┐                   ┌──────────┐                      │
│            │ TcpPool  │◄─── msgpack ────► │ TcpPool  │                      │
│            └────┬─────┘                   └────┬─────┘                      │
└─────────────────┼──────────────────────────────┼────────────────────────────┘
                  │ TCP :6789                    │
┌─────────────────┼──────────────────────────────┼────────────────────────────┐
│                 ▼           SERVER             ▼                            │
│          ┌───────────┐                  ┌───────────┐                       │
│          │ TcpServer │                  │ TcpServer │                       │
│          └─────┬─────┘                  └─────┬─────┘                       │
│                │                              │                             │
│                ▼                              ▼                             │
│   ┌────────────────────────────────────────────────────────────┐           │
│   │                    QueueManager                             │           │
│   │  ┌─────────────────────────────────────────────────────┐   │           │
│   │  │              N Shards (auto-detected)                │   │           │
│   │  │  ┌─────────┬─────────┬─────────┬─────────┐          │   │           │
│   │  │  │ Shard 0 │ Shard 1 │   ...   │ Shard N │          │   │           │
│   │  │  │┌───────┐│┌───────┐│         │┌───────┐│          │   │           │
│   │  │  ││PQueue ││PQueue ││         ││PQueue ││          │   │           │
│   │  │  │└───────┘│└───────┘│         │└───────┘│          │   │           │
│   │  │  └─────────┴─────────┴─────────┴─────────┘          │   │           │
│   │  └─────────────────────────────────────────────────────┘   │           │
│   │                           │                                 │           │
│   │  ┌────────────────────────┼────────────────────────────┐   │           │
│   │  │  jobIndex (Map)        │   completedJobs (Set)      │   │           │
│   │  │  customIdMap (LRU)     │   jobResults (LRU)         │   │           │
│   │  └────────────────────────┼────────────────────────────┘   │           │
│   └───────────────────────────┼─────────────────────────────────┘           │
│                               │                                             │
│   ┌───────────────────────────┼─────────────────────────────────┐           │
│   │                           ▼                                 │           │
│   │  ┌─────────────┐    ┌──────────┐    ┌─────────────┐        │           │
│   │  │ WriteBuffer │───►│ SQLite   │◄───│ ReadThrough │        │           │
│   │  │ (10ms batch)│    │ WAL Mode │    │   Cache     │        │           │
│   │  └─────────────┘    └──────────┘    └─────────────┘        │           │
│   │                      Persistence                            │           │
│   └─────────────────────────────────────────────────────────────┘           │
│                                                                             │
│   ┌───────────────────────────────────────────────────────────┐             │
│   │  Background Tasks                                          │             │
│   │  • Scheduler (cron, delayed jobs)    • Stall detector     │             │
│   │  • DLQ maintenance (retry, expire)   • Lock expiration    │             │
│   │  • Cleanup (memory bounds)           • S3 backup          │             │
│   └───────────────────────────────────────────────────────────┘             │
└─────────────────────────────────────────────────────────────────────────────┘
```

**Request Flow:**

1. **PUSH**: Client → TcpPool → TcpServer → QueueManager → Shard[hash(queue)] → PriorityQueue → WriteBuffer → SQLite
2. **PULL**: Client → TcpServer → QueueManager → Shard → PriorityQueue.pop() → Job (state: active)
3. **ACK**: Client → TcpServer → AckBatcher → Shard.complete() → jobResults (LRU) + completedJobs (Set)
4. **FAIL**: Client → TcpServer → Shard.fail() → retry (backoff) OR → DLQ (max attempts)

## Directory Structure

```
src/
├── cli/              # CLI interface (commands/, client.ts, output.ts)
├── client/           # Embedded SDK (Queue, Worker, FlowProducer, QueueGroup)
│   ├── queue/        # Queue with DLQ, stall detection
│   ├── worker/       # Worker with heartbeat, ack batching
│   └── tcp/          # Connection pool, reconnection
├── domain/           # Pure business logic (types/, queue/)
│   └── queue/        # Shard, PriorityQueue, DlqShard, UniqueKeyManager
├── application/      # Use cases (operations/, managers)
│   ├── operations/   # push, pull, ack, query, queueControl
│   └── *Manager.ts   # DLQ, Events, Workers, JobLogs, Stats
├── infrastructure/   # External (persistence/, server/, scheduler/, backup/)
└── shared/           # Utilities (hash, lock, lru, skipList, minHeap)
```

## Code Guidelines

- **MAX 300 lines per file** - split if larger
- One concern per file (Single Responsibility)
- Export only what's needed

## Sharding (auto-detected from CPU cores)

Shard count is automatically calculated based on CPU cores (power of 2, max 64):

```typescript
// Calculated at startup based on navigator.hardwareConcurrency
// Examples: 4 cores → 4 shards, 10 cores → 16 shards, 20 cores → 32 shards
const SHARD_COUNT = calculateShardCount(); // Power of 2, capped at 64
const SHARD_MASK = SHARD_COUNT - 1;
const shardIndex = (key: string) => fnv1aHash(key) & SHARD_MASK;
```

## Lock Hierarchy (acquire in order)

1. `jobIndex` → 2. `completedJobs` → 3. `shards[N]` → 4. `processingShards[N]`

```typescript
// CORRECT: read first, then acquire lock
const completed = completedJobs.has(id);
const shard = await shards[idx].acquire();
try {
  /* work */
} finally {
  shard.release();
}
```

## Memory Bounds

| Collection    | Max Size | Eviction   |
| ------------- | -------- | ---------- |
| completedJobs | 50,000   | FIFO batch |
| jobResults    | 5,000    | LRU        |
| jobLogs       | 10,000   | LRU        |
| customIdMap   | 50,000   | LRU        |

Cleanup runs every 10s. Evicts 10% when full.

## Environment Variables

```bash
# Server
TCP_PORT=6789              HTTP_PORT=6790
HOST=0.0.0.0               BUNQUEUE_DATA_PATH=./data/bunq.db
AUTH_TOKENS=token1,token2  CORS_ALLOW_ORIGIN=*
METRICS_AUTH=false          # Require auth for /metrics endpoint
TCP_SOCKET_PATH=           # Unix socket for TCP (overrides host/port)
HTTP_SOCKET_PATH=          # Unix socket for HTTP

# Data path (priority: BUNQUEUE_DATA_PATH > BQ_DATA_PATH > DATA_PATH > SQLITE_PATH)
# Or set programmatically: new Queue('q', { embedded: true, dataPath: './data/q.db' })

# S3 Backup
S3_BACKUP_ENABLED=0        S3_BUCKET=my-bucket
S3_ACCESS_KEY_ID=          S3_SECRET_ACCESS_KEY=
S3_REGION=us-east-1        S3_ENDPOINT=
S3_BACKUP_INTERVAL=21600000  S3_BACKUP_RETENTION=7

# Timeouts
SHUTDOWN_TIMEOUT_MS=30000  STATS_INTERVAL_MS=300000
WORKER_TIMEOUT_MS=30000    LOCK_TIMEOUT_MS=5000
WEBHOOK_MAX_RETRIES=3      WEBHOOK_RETRY_DELAY_MS=1000

# Cloud (bunqueue.io dashboard)
BUNQUEUE_CLOUD_URL=        BUNQUEUE_CLOUD_API_KEY=
BUNQUEUE_CLOUD_INSTANCE_NAME=  # defaults to hostname
BUNQUEUE_CLOUD_INTERVAL_MS=5000
BUNQUEUE_CLOUD_USE_WEBSOCKET=true
BUNQUEUE_CLOUD_USE_HTTP=true
BUNQUEUE_CLOUD_REMOTE_COMMANDS=false
BUNQUEUE_CLOUD_INCLUDE_JOB_DATA=false
BUNQUEUE_CLOUD_REDACT_FIELDS=  # comma-separated
BUNQUEUE_CLOUD_EVENTS=         # event filter, comma-separated
BUNQUEUE_CLOUD_BUFFER_SIZE=720
BUNQUEUE_CLOUD_CIRCUIT_BREAKER_THRESHOLD=5
BUNQUEUE_CLOUD_CIRCUIT_BREAKER_RESET_MS=60000
BUNQUEUE_CLOUD_SIGNING_SECRET=  # HMAC signing
```

## TCP Protocol Commands

**Core:** `PUSH`, `PUSHB`, `PULL`, `PULLB`, `ACK`, `ACKB`, `FAIL`

**Query:** `GetJob`, `GetState`, `GetResult`, `GetJobs`, `GetJobCounts`, `GetProgress`, `Count`

**Control:** `Pause`, `Resume`, `Drain`, `Obliterate`, `Clean`, `Cancel`, `Promote`, `Update`, `ChangePriority`

**DLQ:** `Dlq`, `RetryDlq`, `PurgeDlq`

**Cron:** `Cron`, `CronDelete`, `CronList`

**Monitor:** `Stats`, `Metrics`, `Prometheus`, `Ping`, `Heartbeat`, `JobHeartbeat`

**Workers:** `RegisterWorker`, `UnregisterWorker`, `ListWorkers`

**Webhooks:** `AddWebhook`, `RemoveWebhook`, `ListWebhooks`

**Rate:** `RateLimit`, `RateLimitClear`, `SetConcurrency`, `ClearConcurrency`

## CLI Usage

```bash
# Server
bunqueue start --tcp-port 6789 --data-path ./data/queue.db

# Client
bunqueue push <queue> <json> [--priority N] [--delay ms]
bunqueue pull <queue> [--timeout ms]
bunqueue ack <id> [--result json]
bunqueue fail <id> [--error msg]
bunqueue job get|state|cancel|promote|discard <id>
bunqueue queue pause|resume|drain|obliterate <queue>
bunqueue dlq list|retry|purge <queue>
bunqueue cron list|add|delete
bunqueue stats|metrics|health
```

## Simple Mode (Bunqueue)

All-in-one Queue + Worker with routes, middleware, and cron:

```typescript
import { Bunqueue } from 'bunqueue/client';

const app = new Bunqueue('tasks', {
  embedded: true,
  // Single processor OR named routes
  routes: {
    'send-email': async (job) => { return { sent: true }; },
    'send-sms': async (job) => { return { sent: true }; },
  },
  concurrency: 10,
});

// Middleware (onion model: mw1 → mw2 → processor → mw2 → mw1)
app.use(async (job, next) => {
  const start = Date.now();
  const result = await next();
  console.log(`${job.name} took ${Date.now() - start}ms`);
  return result;
});

// Cron
await app.cron('daily', '0 9 * * *', { type: 'report' });
await app.every('ping', 30000, { type: 'health' });

// Events, add, control
app.on('completed', (job, result) => {});
await app.add('send-email', { to: 'user@test.com' });
app.pause(); app.resume();
await app.close();
```

Key: `Bunqueue` = wrapper around `Queue` + `Worker`. Use `processor` OR `routes`, not both. Middleware wraps the processor. Cron delegates to `queue.upsertJobScheduler()`. For distributed (separate producer/consumer), use `Queue` + `Worker` directly.

Source: `src/client/bunqueue.ts`

## Workflow Engine

Multi-step orchestration with saga compensation, branching, parallel steps, retry, nested workflows, loops (doUntil/doWhile), forEach, map, schema validation, subscribe, observability, and human-in-the-loop signals:

```typescript
import { Workflow, Engine } from 'bunqueue/workflow';

const flow = new Workflow('order')
  .step('validate', async (ctx) => {
    const { orderId } = ctx.input as { orderId: string };
    return { orderId };
  })
  .step('charge', async (ctx) => {
    return { txId: 'tx_123' };
  }, { compensate: async () => { /* auto-rollback on failure */ }, retry: 3 })
  .parallel((w) => w
    .step('notify-warehouse', async () => ({ notified: true }))
    .step('notify-email', async () => ({ sent: true }))
  )
  .branch((ctx) => (ctx.steps['classify'] as { tier: string }).tier)
  .path('vip', (w) => w.step('vip-handler', async () => ({ discount: 20 })))
  .path('basic', (w) => w.step('basic-handler', async () => ({ discount: 0 })))
  .forEach(                                        // iterate over items
    (ctx) => (ctx.input as { items: unknown[] }).items,
    'process-item', async (ctx) => { return { item: ctx.steps.__item }; },
  )
  .map('summary', (ctx) => ({ total: 42 }))       // synchronous transform
  .doUntil(                                         // loop until condition
    (ctx) => (ctx.steps['poll'] as any)?.ready,
    (w) => w.step('poll', async () => ({ ready: true })),
    { maxIterations: 10 },
  )
  .waitFor('approval', { timeout: 86400000 })      // pauses until engine.signal(), 24h timeout
  .subWorkflow('payment', (ctx) => ({ amount: 99 }))  // nested workflow
  .step('finalize', async (ctx) => {
    const decision = ctx.signals['approval'];
    const payment = ctx.steps['sub:payment'];
    return { done: true };
  });

const engine = new Engine({ embedded: true, onEvent: (e) => console.log(e.type) });
engine.register(flow);
engine.on('step:retry', (e) => logger.warn(e));
const run = await engine.start('order', { orderId: 'ORD-1' });
await engine.signal(run.id, 'approval', { approved: true });
engine.archive(7 * 24 * 60 * 60 * 1000); // archive old executions
const unsub = engine.subscribe(run.id, (e) => console.log(e.type)); // per-execution subscribe
```

Key: `Workflow` = pure DSL builder (step/branch/path/parallel/waitFor/subWorkflow/doUntil/doWhile/forEach/map). `Engine` = facade over Queue + Worker + SQLite Store + Executor + Emitter. Each step runs as a bunqueue job on `__wf:steps` queue. Features: retry with exponential backoff, parallel via Promise.allSettled, nested workflows via polling, signal timeout, loops (doUntil/doWhile/forEach), map transforms, schema validation (duck-typed .parse()), per-execution subscribe, typed events (11 types), cleanup/archival. Compensation runs in reverse order on failure.

Source: `src/client/workflow/` (workflow.ts, engine.ts, executor.ts, store.ts, runner.ts, emitter.ts, loops.ts, types.ts)

## Client SDK

```typescript
import { Queue, Worker } from 'bunqueue/client';

// Queue (TCP mode)
const queue = new Queue<T>('emails', { connection: { port: 6789 } });

// Queue (embedded mode — programmatic dataPath, no env var needed)
const embeddedQueue = new Queue<T>('emails', {
  embedded: true,
  dataPath: './data/myapp.db',
});
await queue.add('send', { email: 'user@test.com' });
await queue.add('payment', data, { durable: true }); // Immediate disk write
queue.pause();
queue.resume();
queue.drain();
queue.obliterate();

// Worker
const worker = new Worker(
  'emails',
  async (job) => {
    await job.updateProgress(50);
    return { sent: true };
  },
  { concurrency: 5, heartbeatInterval: 10000 }
);

worker.on('completed', (job, result) => {});
worker.on('failed', (job, err) => {});

// Stall Detection (embedded only)
queue.setStallConfig({ stallInterval: 30000, maxStalls: 3, gracePeriod: 5000 });

// DLQ (embedded only)
queue.setDlqConfig({ autoRetry: true, maxAge: 604800000, maxEntries: 10000 });
const entries = queue.getDlq({ reason: 'timeout' });
queue.retryDlq();
queue.purgeDlq();

// Auto-batching (TCP mode, enabled by default)
// Transparently batches concurrent add() calls into PUSHB commands.
// Strategy: flush immediately if idle, buffer during in-flight flush.
// Sequential await has zero overhead; concurrent adds get ~3x speedup.
const queue2 = new Queue('jobs', {
  autoBatch: { maxSize: 50, maxDelayMs: 5 },  // defaults
});
// Sequential: no penalty, each add() sends immediately
for (const item of items) {
  await queue2.add('task', item); // same speed as without batching
}
// Concurrent: adds batch into a single PUSHB round-trip (~3x faster)
await Promise.all([
  queue2.add('a', { x: 1 }),
  queue2.add('b', { x: 2 }),
  queue2.add('c', { x: 3 }),
]);
// Durable jobs bypass the batcher (sent as individual PUSH):
await queue2.add('critical', data, { durable: true });
// Disable auto-batching:
const queue3 = new Queue('jobs', { autoBatch: { enabled: false } });
```

## Job Options

```typescript
interface JobOptions {
  priority?: number; // Higher = sooner
  delay?: number; // ms before processing
  attempts?: number; // Max retries (default: 3)
  backoff?: number; // Retry backoff (default: 1000ms)
  timeout?: number; // Processing timeout
  jobId?: string; // Custom ID (idempotent)
  removeOnComplete?: boolean;
  removeOnFail?: boolean;
  durable?: boolean; // Bypass write buffer
}
```

## Worker Options

```typescript
interface WorkerOptions {
  concurrency?: number; // Parallel jobs (default: 1)
  heartbeatInterval?: number; // Stall detection (default: 10000, 0=disabled)
  batchSize?: number; // Pull batch (default: 10, max: 1000)
  pollTimeout?: number; // Long poll (default: 0, max: 30000)
  useLocks?: boolean; // Lock-based ownership (default: true)
}
```

## SQLite Schema (Key Tables)

```sql
-- Jobs: id, queue, data, priority, state, run_at, attempts, ...
CREATE INDEX idx_jobs_queue_state ON jobs(queue, state);
CREATE INDEX idx_jobs_run_at ON jobs(run_at) WHERE state IN ('waiting','delayed');

-- DLQ: id, job_id, queue, entry (msgpack blob), entered_at
-- Cron: name, queue, data, schedule, repeat_every, next_run, timezone
-- Results: job_id, result, completed_at
```

## Testing

```bash
bun test                           # All tests (~5065 tests)
bun scripts/tcp/run-all-tests.ts   # TCP tests (~52 suites)
bun run bench                      # Benchmarks
```

## Publishing

Always use `bun publish` (not `npm publish`) to publish to npm.

```bash
bun publish
```

## Performance

| Mode               | Throughput      | Data Loss Risk |
| ------------------ | --------------- | -------------- |
| Buffered (default) | ~100k jobs/sec  | Up to 10ms     |
| Durable            | ~10k jobs/sec   | None           |
| Auto-batch (TCP)   | ~145k ops/s (concurrent), ~10k ops/s (sequential) | None (same as PUSH/PUSHB) |

## Debug Endpoints

```bash
curl http://localhost:6790/health     # Health + memory
curl http://localhost:6790/heapstats  # Object breakdown
curl -X POST http://localhost:6790/gc # Force GC
```

## Memory Debugging

```typescript
import { heapStats } from 'bun:jsc';
Bun.gc(true);
const stats = heapStats();
console.log(stats.objectCount, stats.objectTypeCounts);

// Check internal collections
const mem = queueManager.getMemoryStats();
// jobIndex, completedJobs, processingTotal, queuedTotal, temporalIndexTotal
```

## Background Tasks

| Task            | Interval | Purpose                        |
| --------------- | -------- | ------------------------------ |
| Cleanup         | 10s      | Memory cleanup, orphan removal |
| Stall check     | 5s       | Detect unresponsive jobs       |
| Dependency      | 100ms    | Process job dependencies       |
| DLQ maintenance | 60s      | Auto-retry, expiration         |
| Lock expiration | 5s       | Remove expired locks           |
