Five clients. One mental model.

npm install queen-mq

import { Queen } from 'queen-mq'

// 1) Single server
const queen = new Queen('http://localhost:6632')

// 2) Multiple servers, automatic failover + load balancing
const queenHA = new Queen([
  'http://queen-a:6632',
  'http://queen-b:6632',
])

// 3) Full config (JWT, custom headers, retry, affinity)
const queenFull = new Queen({
  urls: ['http://queen-a:6632', 'http://queen-b:6632'],
  loadBalancingStrategy: 'affinity',  // 'round-robin' | 'session' | 'affinity'
  enableFailover: true,
  retryAttempts: 3,
  bearerToken: process.env.QUEEN_TOKEN,
})

await queen.queue('orders')
  .config({
    leaseTime: 300,
    retryLimit: 3,
    retentionSeconds: 86400,
    encryptionEnabled: false,
  })
  .create()

// Plain push
await queen.queue('orders').push([
  { data: { orderId: 1, amount: 99.99 } },
])

// Push to a partition for ordering
await queen.queue('orders').partition('user-123').push([
  { data: { orderId: 2, amount: 50 } },
])

// With a transactionId for exactly-once dedup
await queen.queue('orders').push([
  { transactionId: 'order-2-v1', data: { orderId: 2, amount: 50 } },
])

// Many at once
await queen.queue('orders').push([
  { data: { orderId: 3 } },
  { data: { orderId: 4 } },
  { data: { orderId: 5 } },
])

const messages = await queen.queue('orders').batch(10).pop()

for (const m of messages) {
  try {
    await processOrder(m.data)
    await queen.ack(m, true)             // success
  } catch (e) {
    await queen.ack(m, false, { error: e.message })  // retry
  }
}

await queen.queue('orders')
  .concurrency(10)        // 10 parallel workers
  .batch(20)              // up to 20 messages per poll
  .each()                 // handler receives one message at a time
  .consume(async (m) => {
    await processOrder(m.data)
    // auto-ack on resolve, auto-nack on throw
  })

// One round-trip drains up to 200 messages spread across up to 50 partitions.
// batch(200) is the GLOBAL cap; partitions(50) is the partition fan-out cap.
const messages = await queen.queue('events')
  .batch(200)
  .partitions(50)
  .wait(true)
  .pop()

for (const m of messages) {
  console.log(`from ${m.partition}:`, m.data)   // m.partitionId / m.leaseId baked in
}

await queen.renew(messages)   // one renew extends every claimed partition's lease

// Same builder works for long-running consumers:
await queen.queue('events')
  .batch(100)
  .partitions(8)
  .consume(async (msgs) => {
    // msgs may span multiple partitions; FIFO is preserved within each partition
    for (const m of msgs) await process(m.data)
  })

// Group A: emailer (load-balanced across two workers)
queen.queue('events').group('emailer').consume(emailHandler1)
queen.queue('events').group('emailer').consume(emailHandler2)

// Group B: analytics, sees the same messages independently
queen.queue('events').group('analytics').consume(analyticsHandler)

// Skip backlog, only process messages pushed after this group joins
queen.queue('events').group('realtime')
  .subscriptionMode('new')
  .consume(rtHandler)

// Replay from a timestamp
queen.queue('events').group('replay-2024')
  .subscriptionFrom('2024-01-01T00:00:00Z')
  .consume(replayHandler)

await queen
  .transaction()
  .ack(message, 'completed', { consumerGroup: 'analytics' })
  .queue('next-stage').partition('p2').push([
    { data: derived, transactionId: message.transactionId + '-next' },
  ])
  .commit()

// Server holds the connection until a message arrives or 30s elapses.
const messages = await queen.queue('jobs')
  .batch(10)
  .wait(true)
  .pop()

await queen.queue('long-jobs')
  .renewLease(true, 2000)   // renew every 2 seconds
  .autoAck(false)
  .each()
  .consume(async (m) => {
    await runLongJob(m.data)   // can take minutes
  })

await queen.queue('events')
  .buffer({ messageCount: 100, timeMillis: 100 })
  .push([{ data: event }])

// Flush at shutdown
await queen.flushAllBuffers()

const dlq = await queen.queue('orders').dlq().limit(50).get()
console.log('failed:', dlq.total)
for (const m of dlq.messages) {
  console.log(m.errorMessage, m.data)
}