Remove PullResult wrapper from queue adapter API

pull() now returns Promise<T | undefined> instead of
Promise<PullResult<T> | undefined>. Also fixes README to match the
actual createQueue(handler, adapter, options?) signature.

Author: brielov

.changeset/remove-pull-result-wrapper.md

@@ -0,0 +1,5 @@
+---
+"@suckless/queue": minor
+---
+
+Remove PullResult wrapper from queue adapter API. `pull` now returns `Promise<T | undefined>` instead of `Promise<PullResult<T> | undefined>`, matching the same simplification applied to @suckless/cache.

packages/queue/README.md

@@ -11,13 +11,14 @@ npm install @suckless/queue
 ## Usage
 
 ```ts
-import { createQueue } from "@suckless/queue"
+import { createQueue, memoryAdapter } from "@suckless/queue"
 
 const queue = createQueue<string>(
 	async (url) => {
 		const res = await fetch(url)
 		await saveToDb(res)
 	},
+	memoryAdapter(),
 	{ concurrency: 4 },
 )
 
@@ -37,13 +38,14 @@ Handler errors are caught to prevent worker death. The handler is responsible fo
 
 ## API
 
-### `createQueue<T>(handler, options?): Queue<T>`
+### `createQueue<T>(handler, adapter, options?): Queue<T>`
 
 Creates a new queue. Workers start pulling immediately.
 
 - `handler` — called for each item. May be sync or async.
-- `options.adapter` — storage backend. Defaults to `memoryAdapter()`.
+- `adapter` — a `QueueAdapter<T>` storage backend.
 - `options.concurrency` — max concurrent handlers. Must be a positive finite integer. Defaults to `1`.
+- `options.onError` — called when the handler throws. Receives the error and the item.
 
 ### `queue.push(item): Promise<void>`
 
@@ -62,7 +64,9 @@ Number of handlers currently executing.
 The queue implements `AsyncDisposable`. Disposing marks the queue as closed, stops workers from pulling new items, waits for in-flight handlers, then disposes the adapter:
 
 ```ts
-await using queue = createQueue<Job>(processJob, { concurrency: 4 })
+await using queue = createQueue<Job>(processJob, memoryAdapter(), {
+	concurrency: 4,
+})
 ```
 
 ## Adapters
@@ -77,7 +81,7 @@ Implement `QueueAdapter<T>` to plug in any backend:
 
 ```ts
 import { RedisClient } from "bun"
-import type { PullResult, QueueAdapter } from "@suckless/queue"
+import type { QueueAdapter } from "@suckless/queue"
 
 function redisAdapter<T>(key: string, url?: string): QueueAdapter<T> {
 	const redis = new RedisClient(url)
@@ -90,7 +94,7 @@ function redisAdapter<T>(key: string, url?: string): QueueAdapter<T> {
 			while (!signal.aborted) {
 				const result = await redis.send("BRPOP", [key, "1"])
 				if (result) {
-					return { value: JSON.parse(result[1]) } as PullResult<T>
+					return JSON.parse(result[1]) as T
 				}
 			}
 			return undefined

packages/queue/src/index.test.ts

@@ -1,10 +1,5 @@
 import { describe, expect, mock, test } from "bun:test"
-import {
-	type PullResult,
-	type QueueAdapter,
-	createQueue,
-	memoryAdapter,
-} from "."
+import { createQueue, type QueueAdapter, memoryAdapter } from "."
 
 function delay(ms: number): Promise<void> {
 	return new Promise<void>((resolve) => {
@@ -20,9 +15,9 @@ describe("memoryAdapter", () => {
 		await adapter.push("c")
 
 		const signal = AbortSignal.abort()
-		expect(await adapter.pull(signal)).toEqual({ value: "a" })
-		expect(await adapter.pull(signal)).toEqual({ value: "b" })
-		expect(await adapter.pull(signal)).toEqual({ value: "c" })
+		expect(await adapter.pull(signal)).toBe("a")
+		expect(await adapter.pull(signal)).toBe("b")
+		expect(await adapter.pull(signal)).toBe("c")
 	})
 
 	test("blocking pull resolves when item pushed", async () => {
@@ -32,7 +27,7 @@ describe("memoryAdapter", () => {
 		const pulling = adapter.pull(controller.signal)
 		await adapter.push(42)
 
-		expect(await pulling).toEqual({ value: 42 })
+		expect(await pulling).toBe(42)
 		controller.abort()
 	})
 
@@ -43,7 +38,7 @@ describe("memoryAdapter", () => {
 		const pulling = adapter.pull(controller.signal)
 		await adapter.push("direct")
 
-		expect(await pulling).toEqual({ value: "direct" })
+		expect(await pulling).toBe("direct")
 		controller.abort()
 	})
 
@@ -73,8 +68,8 @@ describe("memoryAdapter", () => {
 		await adapter.push(2)
 
 		// Drain the buffer
-		expect(await adapter.pull(controller.signal)).toEqual({ value: 1 })
-		expect(await adapter.pull(controller.signal)).toEqual({ value: 2 })
+		expect(await adapter.pull(controller.signal)).toBe(1)
+		expect(await adapter.pull(controller.signal)).toBe(2)
 
 		// Now buffer is empty — this pull will block
 		const pulling = adapter.pull(controller.signal)
@@ -107,9 +102,9 @@ describe("memoryAdapter", () => {
 		await adapter.push("b")
 		await adapter.push("c")
 
-		expect(await pull1).toEqual({ value: "a" })
-		expect(await pull2).toEqual({ value: "b" })
-		expect(await pull3).toEqual({ value: "c" })
+		expect(await pull1).toBe("a")
+		expect(await pull2).toBe("b")
+		expect(await pull3).toBe("c")
 		controller.abort()
 	})
 
@@ -121,27 +116,9 @@ describe("memoryAdapter", () => {
 		await adapter.push("")
 
 		const signal = AbortSignal.abort()
-		expect(await adapter.pull(signal)).toEqual({ value: 0 })
-		expect(await adapter.pull(signal)).toEqual({ value: false })
-		expect(await adapter.pull(signal)).toEqual({ value: "" })
-	})
-
-	test("undefined items pass through correctly", async () => {
-		const adapter = memoryAdapter<undefined>()
-
-		await adapter.push(undefined)
-		await adapter.push(undefined)
-
-		const signal = AbortSignal.abort()
-		const result = await adapter.pull(signal)
-		expect(result).toEqual({ value: undefined })
-		expect(result).not.toBeUndefined()
-
-		// Second pull also returns wrapped undefined, not bare undefined
-		expect(await adapter.pull(signal)).toEqual({ value: undefined })
-
-		// Empty buffer with aborted signal returns bare undefined
-		expect(await adapter.pull(signal)).toBeUndefined()
+		expect(await adapter.pull(signal)).toBe(0)
+		expect(await adapter.pull(signal)).toBe(false)
+		expect(await adapter.pull(signal)).toBe("")
 	})
 })
 
@@ -404,21 +381,6 @@ describe("createQueue", () => {
 		await queue[Symbol.asyncDispose]()
 	})
 
-	test("undefined items pass through handler", async () => {
-		let count = 0
-		const queue = createQueue<undefined>(() => {
-			count++
-		}, memoryAdapter())
-
-		await queue.push(undefined)
-		await queue.push(undefined)
-		await queue.push(undefined)
-		await queue.drain()
-
-		expect(count).toBe(3)
-		await queue[Symbol.asyncDispose]()
-	})
-
 	test("multiple concurrent drain waiters all resolve", async () => {
 		const queue = createQueue<number>(async () => {
 			await delay(10)
@@ -476,7 +438,7 @@ describe("createQueue with spy adapter", () => {
 		const adapter: QueueAdapter<number> = {
 			push: pushFn,
 			pull: () =>
-				new Promise<PullResult<number> | undefined>((resolve) => {
+				new Promise<number | undefined>((resolve) => {
 					setTimeout(() => {
 						resolve(undefined)
 					}, 100)
@@ -493,7 +455,7 @@ describe("createQueue with spy adapter", () => {
 
 	test("workers call adapter pull", async () => {
 		const pullFn = mock((_signal: AbortSignal) =>
-			Promise.resolve<PullResult<number> | undefined>(undefined),
+			Promise.resolve<number | undefined>(undefined),
 		)
 		const adapter: QueueAdapter<number> = {
 			push: () => Promise.resolve(),
@@ -513,7 +475,7 @@ describe("createQueue with spy adapter", () => {
 		const adapter: QueueAdapter<number> = {
 			push: () => Promise.reject(new Error("push failed")),
 			pull: () =>
-				new Promise<PullResult<number> | undefined>((resolve) => {
+				new Promise<number | undefined>((resolve) => {
 					setTimeout(() => {
 						resolve(undefined)
 					}, 100)
@@ -554,7 +516,7 @@ describe("createQueue with spy adapter", () => {
 		const adapter: QueueAdapter<number> = {
 			push: () => Promise.resolve(),
 			pull: () =>
-				new Promise<PullResult<number> | undefined>((resolve) => {
+				new Promise<number | undefined>((resolve) => {
 					setTimeout(() => {
 						resolve(undefined)
 					}, 100)

packages/queue/src/index.ts

@@ -1,12 +1,7 @@
-/** Wrapper that distinguishes a pulled value from "no value available." */
-export interface PullResult<T> {
-	value: T
-}
-
 /** Storage backend for queue items. */
 export interface QueueAdapter<T> extends AsyncDisposable {
 	push(item: T): Promise<void>
-	pull(signal: AbortSignal): Promise<PullResult<T> | undefined>
+	pull(signal: AbortSignal): Promise<T | undefined>
 }
 
 /** Producer/consumer queue with pluggable storage. */
@@ -19,13 +14,13 @@ export interface Queue<T> extends AsyncDisposable {
 /** In-memory FIFO queue adapter with blocking pull. */
 export function memoryAdapter<T>(): QueueAdapter<T> {
 	const buffer: T[] = []
-	const waiters: ((result: PullResult<T> | undefined) => void)[] = []
+	const waiters: ((item: T | undefined) => void)[] = []
 
 	return {
 		push(item) {
 			const waiter = waiters.shift()
 			if (waiter) {
-				waiter({ value: item })
+				waiter(item)
 			} else {
 				buffer.push(item)
 			}
@@ -34,18 +29,16 @@ export function memoryAdapter<T>(): QueueAdapter<T> {
 
 		pull(signal) {
 			if (buffer.length > 0) {
-				return Promise.resolve<PullResult<T>>({
-					// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- length check guarantees shift returns T
-					value: buffer.shift() as T,
-				})
+				// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- length check guarantees shift returns T
+				return Promise.resolve(buffer.shift() as T)
 			}
 			if (signal.aborted) {
 				return Promise.resolve(undefined)
 			}
-			return new Promise<PullResult<T> | undefined>((resolve) => {
-				const waiter = (result: PullResult<T> | undefined) => {
+			return new Promise<T | undefined>((resolve) => {
+				const waiter = (item: T | undefined) => {
 					signal.removeEventListener("abort", onAbort)
-					resolve(result)
+					resolve(item)
 				}
 				const onAbort = () => {
 					const idx = waiters.indexOf(waiter)
@@ -104,27 +97,27 @@ export function createQueue<T>(
 
 	async function worker(): Promise<void> {
 		while (!controller.signal.aborted) {
-			let result: PullResult<T> | undefined
+			let item: T | undefined
 			try {
 				// oxlint-disable-next-line no-await-in-loop -- worker loop is sequential by design
-				result = await adapter.pull(controller.signal)
+				item = await adapter.pull(controller.signal)
 			} catch {
 				// Adapter error — exit gracefully to avoid unhandled rejections.
 				break
 			}
-			if (result === undefined) {
+			if (item === undefined) {
 				break
 			}
 			running++
 			try {
-				await handler(result.value) // oxlint-disable-line no-await-in-loop
+				await handler(item) // oxlint-disable-line no-await-in-loop
 			} catch (error) {
 				try {
 					onError?.(
 						error instanceof Error
 							? error
 							: new Error(String(error), { cause: error }),
-						result.value,
+						item,
 					)
 				} catch {
 					// Prevent onError failures from killing the worker.