feat: event listeners on Worker, Queue, and Job

README.md

@@ -204,6 +204,8 @@ In-repo: [`docs/engine.md`](docs/engine.md) (engine internals), [`docs/history.m
 | Idempotent produce (`IDMP`) | ✓ | — | — | — |
 | Stable job IDs (`addUnique`) | ✓ | ✓ | ✓ | — |
 | Result backends (`getJobResult` / `waitForResult`) | ✓ | ✓ | ✓ | — |
+| Event-driven completion wait (`waitUntilFinished`) | ✓ | ✓ | ✓ | — |
+| Worker / Queue / Job event listeners (`EventEmitter`-style) | ✓ | ✓ | ✓ | partial |
 | Delayed jobs | ✓ | ✓ | ✓ | — |
 | Idempotent delayed scheduling | ✓ | — | — | — |
 | Cancel scheduled job | ✓ | ✓ | ✓ | — |

chasquimq-node/README.md

@@ -69,9 +69,9 @@ main()
 | Surface | What it does |
 |---|---|
 | `Queue<DataType, ResultType, NameType>` | Producer + queue inspection. `add` / `addBulk` / `addUnique` / `getJob` / `getJobs` / `getJobState` / `getJobCounts` / `getWaitingCount` / `getActiveCount` / `getDelayedCount` / `getCompletedCount` / `getFailedCount` / `count` / `getJobResult` / `peekDlq` / `replayDlq` / `cancelDelayed` / `getRepeatableJobs` / `removeRepeatableByKey` / `pause` / `resume` / `isPaused` / `remove` / `removeReport` / `drain` / `clean` / `obliterate`. `[Symbol.asyncDispose]`. |
-| `Worker<DataType, ResultType, NameType>` | Consumer pool. tokio-side dispatch, opt-in result storage (`storeResults: true`), `EventEmitter` events (`completed` / `failed` / `error`), `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. |
-| `Job<DataType, ResultType, NameType>` | Read-only handle. `id`, `name`, `data`, `attemptsMade`, `waitForResult({ timeoutMs, intervalMs, signal })`. |
-| `QueueEvents` | Cross-process pub/sub via the events stream. Subscribe to `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained`. `[Symbol.asyncDispose]`. |
+| `Worker<DataType, ResultType, NameType>` | Consumer pool. tokio-side dispatch, opt-in result storage (`storeResults: true`), `EventEmitter` events (`ready` / `active` / `completed` / `failed` / `error` / `closing` / `closed` / `drained` / `paused` / `resumed`), `pause` / `resume` / `isPaused`. `[Symbol.asyncDispose]`. |
+| `Job<DataType, ResultType, NameType>` | Read-only handle. `id`, `name`, `data`, `attemptsMade`, `waitForResult({ timeoutMs, intervalMs, signal })`, `waitUntilFinished(queueEvents, ttl?)`. |
+| `QueueEvents` | Cross-process pub/sub via the events stream. Subscribe to `waiting` / `active` / `completed` / `failed` / `dlq` / `retry-scheduled` / `delayed` / `drained` / `retries-exhausted`, plus per-id channels (`completed:<jobId>` / `failed:<jobId>` / `active:<jobId>`) for targeted subscribers. `[Symbol.asyncDispose]`. |
 | `BackoffSpec` | Builders: `.fixed(delayMs)` / `.exponential(initialMs, { multiplier, maxMs, jitterMs })`. |
 | `UnrecoverableError` | Throw from your handler to bypass retries and route the job directly to DLQ. |
 | `NotSupportedError` | Surfaces from APIs that aren't on the chasquimq roadmap (e.g. parent/child flows). |
@@ -190,6 +190,89 @@ await queue.resume()           // lift it everywhere
 
 The same durable flag is what the CLI's `chasqui pause <queue>` / `chasqui resume <queue>` toggle. Both surfaces are idempotent — double-pause / double-resume are no-ops.
 
+### Subscribing to events
+
+Two layered surfaces: per-worker `EventEmitter` events for in-process observation, and the cross-process `QueueEvents` stream for fan-out across workers, dashboards, and ops tooling.
+
+In-process `Worker` events fire on the local worker only and use plain `EventEmitter` semantics:
+
+```ts
+const worker = new Worker("emails", handler, { connection })
+
+worker.on("ready", () => console.log("engine loop started"))
+worker.on("active", (job) => console.log("running", job.id))
+worker.on("completed", (job, result) => console.log("ok", job.id, result))
+worker.on("failed", (job, err) => console.error("failed", job?.id, err))
+worker.on("error", (err) => console.error("engine error", err))
+worker.on("drained", () => console.log("queue is empty"))
+worker.on("paused", () => console.log("worker parked"))
+worker.on("resumed", () => console.log("worker resumed"))
+worker.on("closing", () => console.log("shutting down"))
+worker.on("closed", () => console.log("shutdown complete"))
+```
+
+`drained` is the only `Worker` event that requires a cross-process subscription (the engine emits it on the events stream, since "no jobs left" is a queue-wide observation). It's lazily wired: the first `worker.on('drained', ...)` call spawns an embedded `QueueEvents` subscriber, torn down on `worker.close()`. Workers that never subscribe pay no extra Redis connections.
+
+Cross-process `QueueEvents` listens on the events stream — every process running a `QueueEvents` instance for the same queue sees every transition, regardless of which worker emitted it:
+
+```ts
+import { QueueEvents } from "chasquimq"
+
+const events = new QueueEvents("emails", { connection })
+await events.waitUntilReady()
+
+events.on("waiting", ({ jobId, name }) => { /* enqueued */ })
+events.on("active", ({ jobId, name, attempt }) => { /* started */ })
+events.on("completed", ({ jobId, name, attempt }) => { /* ok */ })
+events.on("failed", ({ jobId, failedReason, attempt }) => { /* err */ })
+events.on("drained", () => { /* empty */ })
+
+// chasquimq-specific extensions
+events.on("retry-scheduled", ({ jobId, attempt, backoffMs }) => { /* next try */ })
+events.on("dlq", ({ jobId, reason, attempt }) => { /* moved to DLQ */ })
+events.on("retries-exhausted", ({ jobId, attemptsMade, reason }) => { /* DLQ */ })
+
+// Per-id channels — fire only for the named job. Used internally by
+// Job.waitUntilFinished; useful for targeted UI updates without
+// filtering every broadcast event by jobId.
+events.on(`completed:${jobId}`, ({ jobId }) => { /* this job, done */ })
+events.on(`failed:${jobId}`, ({ failedReason }) => { /* this job, failed */ })
+
+await events.close()
+```
+
+`completed` payloads from the events stream do **not** carry the handler's return value (that would double-allocate the payload onto every subscriber). To read the value, pair `QueueEvents` with `Queue.getJobResult(jobId)` and run the worker with `storeResults: true`. The next section's `Job.waitUntilFinished` does this for you.
+
+`progress` and `stalled` listeners are accepted on `Worker` for API stability but are currently no-op — the engine doesn't emit those transitions yet. Attach them safely; they'll start firing when the corresponding engine work lands.
+
+### Awaiting a single job's completion
+
+`Job.waitUntilFinished(queueEvents, ttl?)` is the event-driven completion-wait — subscribes to the per-id `completed:<jobId>` / `failed:<jobId>` channels and resolves / rejects on the first to fire:
+
+```ts
+const events = new QueueEvents("emails", { connection })
+await events.waitUntilReady()
+
+const job = await queue.add("send", { to: "ada@example.com" })
+
+try {
+  // Resolves with the handler's return value (requires
+  // storeResults: true on the worker; otherwise resolves with
+  // undefined). Rejects with new Error(failedReason) on failure.
+  // ttl in ms — omit for an unbounded wait.
+  const result = await job.waitUntilFinished(events, 30_000)
+  console.log("sent:", result)
+} catch (err) {
+  // Either a job failure (engine reason) or a
+  // WaitUntilFinishedTimeoutError (ttl elapsed with no terminal event).
+  console.error(err)
+} finally {
+  await events.close()
+}
+```
+
+Distinct from `Job.waitForResult({ timeoutMs })`: that one polls the `Queue.getJobResult` Redis key and **requires** `storeResults: true` to detect completion at all. `waitUntilFinished` is event-driven, so it detects completion even when no result key was written — but it cannot tell you a job that already finished before the call wired up. Pick `waitUntilFinished` for low-latency awaits of jobs you're about to enqueue; pick `waitForResult` for "did this id ever finish".
+
 ### Inspect jobs
 
 Read-only introspection across the queue's stream, delayed ZSET, DLQ, and result-key surfaces. Bounded scans; no secondary index; zero impact on the producer / consumer hot paths.

chasquimq-node/__test__/event-listeners.test.ts

@@ -0,0 +1,287 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  Queue,
+  QueueEvents,
+  Worker,
+  WaitUntilFinishedTimeoutError,
+} from '../dist/index.js'
+
+const REDIS_URL = process.env.REDIS_URL
+const skipIfNoRedis = REDIS_URL ? describe : describe.skip
+
+skipIfNoRedis('Worker + Queue + Job event listeners', () => {
+  let queueName: string
+  let queue: Queue<{ value: number }, number>
+  let worker: Worker<{ value: number }, number> | undefined
+  let queueEvents: QueueEvents | undefined
+
+  beforeEach(() => {
+    queueName = `qmq-test-evl-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`
+    queue = new Queue(queueName, { connection: parseConn(REDIS_URL!) })
+  })
+
+  afterEach(async () => {
+    if (worker) {
+      await worker.close().catch(() => {})
+      worker = undefined
+    }
+    if (queueEvents) {
+      await queueEvents.close().catch(() => {})
+      queueEvents = undefined
+    }
+    await queue.close().catch(() => {})
+  })
+
+  // --- Worker.on('drained') -------------------------------------------------
+
+  it("Worker emits 'drained' after the engine observes a full→empty transition", async () => {
+    const drainedSpy = vi.fn()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value,
+      {
+        connection: parseConn(REDIS_URL!),
+        concurrency: 4,
+        drainDelay: 200,
+        autorun: false,
+      },
+    )
+    worker.on('drained', drainedSpy)
+    void worker.run()
+
+    // Add then wait for the worker to process the job AND observe an
+    // empty XREADGROUP. The engine emits `drained` on the full→empty
+    // transition only (not on every empty poll).
+    await queue.add('one', { value: 1 })
+
+    await waitFor(() => drainedSpy.mock.calls.length >= 1, 10_000)
+    expect(drainedSpy).toHaveBeenCalled()
+  }, 30_000)
+
+  it("Worker doesn't spawn a drained subscriber when no listener attaches", async () => {
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value,
+      { connection: parseConn(REDIS_URL!), autorun: false },
+    )
+    void worker.run()
+    await queue.add('one', { value: 1 })
+
+    // Give the worker a tick to process, then close. The internal
+    // subscriber field should never have been constructed — observable
+    // by `close()` returning quickly (no extra `XREAD BLOCK` to drain).
+    await new Promise((r) => setTimeout(r, 100))
+
+    const start = Date.now()
+    await worker.close()
+    const elapsed = Date.now() - start
+    // Native consumer shutdown is fast (<1s). A live QueueEvents would
+    // add up to `blockingTimeout` (10s default) wait.
+    expect(elapsed).toBeLessThan(3_000)
+  })
+
+  // --- Worker.on('paused' | 'resumed') -------------------------------------
+
+  it("Worker emits 'paused' and 'resumed' when pause()/resume() are called", async () => {
+    const pausedSpy = vi.fn()
+    const resumedSpy = vi.fn()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value,
+      { connection: parseConn(REDIS_URL!), autorun: false },
+    )
+    worker.on('paused', pausedSpy)
+    worker.on('resumed', resumedSpy)
+    void worker.run()
+    // Let it start
+    await new Promise((r) => setTimeout(r, 50))
+
+    await worker.pause()
+    expect(pausedSpy).toHaveBeenCalledTimes(1)
+
+    worker.resume()
+    expect(resumedSpy).toHaveBeenCalledTimes(1)
+  })
+
+  // --- QueueEvents per-id channels ------------------------------------------
+
+  it("QueueEvents emits per-id `completed:<jobId>` alongside the broadcast", async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value,
+      { connection: parseConn(REDIS_URL!), autorun: false },
+    )
+    void worker.run()
+
+    const job = await queue.add('compute', { value: 42 })
+    const targetedSpy = vi.fn()
+    const broadcastSpy = vi.fn()
+    queueEvents.on(`completed:${job.id}`, targetedSpy)
+    queueEvents.on('completed', broadcastSpy)
+
+    await waitFor(
+      () => targetedSpy.mock.calls.length >= 1 && broadcastSpy.mock.calls.length >= 1,
+      10_000,
+    )
+    const [payload] = targetedSpy.mock.calls[0]!
+    expect((payload as { jobId: string }).jobId).toBe(job.id)
+  }, 30_000)
+
+  it("QueueEvents emits per-id `failed:<jobId>` with the engine reason", async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async () => {
+        throw new Error('boom')
+      },
+      {
+        connection: parseConn(REDIS_URL!),
+        concurrency: 1,
+        maxStalledCount: 1,
+        autorun: false,
+      },
+    )
+    void worker.run()
+
+    const job = await queue.add('fail-me', { value: 0 })
+    const targetedSpy = vi.fn()
+    queueEvents.on(`failed:${job.id}`, targetedSpy)
+
+    await waitFor(() => targetedSpy.mock.calls.length >= 1, 10_000)
+    const [payload] = targetedSpy.mock.calls[0]!
+    expect((payload as { jobId: string; failedReason: string })).toMatchObject({
+      jobId: job.id,
+      failedReason: 'boom',
+    })
+  }, 30_000)
+
+  // --- Job.waitUntilFinished -----------------------------------------------
+
+  it('Job.waitUntilFinished resolves on completed with the stored result', async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value * 3,
+      {
+        connection: parseConn(REDIS_URL!),
+        autorun: false,
+        storeResults: true,
+        resultTtlMs: 60_000,
+      },
+    )
+    void worker.run()
+
+    const job = await queue.add('triple', { value: 7 })
+    const result = await job.waitUntilFinished(queueEvents, 10_000)
+    expect(result).toBe(21)
+  }, 30_000)
+
+  it('Job.waitUntilFinished resolves with undefined when storeResults is off', async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async (job) => job.data.value,
+      { connection: parseConn(REDIS_URL!), autorun: false },
+    )
+    void worker.run()
+
+    const job = await queue.add('no-store', { value: 7 })
+    const result = await job.waitUntilFinished(queueEvents, 10_000)
+    // Event-driven completion still detected; the return value is
+    // unavailable without the result backend.
+    expect(result).toBeUndefined()
+  }, 30_000)
+
+  it('Job.waitUntilFinished rejects with the engine reason on failure', async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    worker = new Worker<{ value: number }, number>(
+      queueName,
+      async () => {
+        throw new Error('handler said no')
+      },
+      {
+        connection: parseConn(REDIS_URL!),
+        concurrency: 1,
+        maxStalledCount: 1,
+        autorun: false,
+      },
+    )
+    void worker.run()
+
+    const job = await queue.add('rejects', { value: 0 })
+    await expect(job.waitUntilFinished(queueEvents, 10_000)).rejects.toThrow(
+      /handler said no/,
+    )
+  }, 30_000)
+
+  it('Job.waitUntilFinished throws WaitUntilFinishedTimeoutError on ttl elapse', async () => {
+    queueEvents = new QueueEvents(queueName, {
+      connection: parseConn(REDIS_URL!),
+      autorun: true,
+      blockingTimeout: 500,
+    })
+    await queueEvents.waitUntilReady()
+
+    // No worker; the job will sit waiting forever.
+    const job = await queue.add('orphan', { value: 0 })
+    await expect(job.waitUntilFinished(queueEvents, 300)).rejects.toThrow(
+      WaitUntilFinishedTimeoutError,
+    )
+  })
+})
+
+async function waitFor(
+  predicate: () => boolean,
+  timeoutMs: number,
+): Promise<void> {
+  const start = Date.now()
+  while (!predicate()) {
+    if (Date.now() - start > timeoutMs)
+      throw new Error(`waitFor timed out after ${timeoutMs}ms`)
+    await new Promise((res) => setTimeout(res, 25))
+  }
+}
+
+function parseConn(url: string) {
+  const u = new URL(url)
+  return {
+    host: u.hostname || '127.0.0.1',
+    port: u.port ? Number(u.port) : 6379,
+    password: u.password || undefined,
+    username: u.username || undefined,
+    db: u.pathname && u.pathname !== '/' ? Number(u.pathname.slice(1)) : undefined,
+  }
+}