@netscript/queue

Provider-agnostic message queue abstraction for NetScript applications. It wraps Fedify battle-tested queue adapters behind a single, unified MessageQueue interface with optional Zod validation and Aspire-based backend auto-discovery. This page is written against the package public surface reported by deno doc. For the full index of packages and plugins return to the reference overview.

The root entrypoint (@netscript/queue) re-exports the factory functions, the port contracts (./ports), the error hierarchy (./errors), and the validation helpers (./validation). The remaining sub-path exports carry the concrete provider adapters and a test double:

• @netscript/queue/ports — the core interfaces and enums.
• @netscript/queue/errors — the queue error classes and codes.
• @netscript/queue/validation — Zod validation helpers.
• @netscript/queue/adapters/deno-kv — Deno KV adapter.
• @netscript/queue/adapters/redis — Redis adapter.
• @netscript/queue/adapters/amqp — RabbitMQ (AMQP) adapter.
• @netscript/queue/adapters/kv-polling — KV-polling adapter for KV Connect.
• @netscript/queue/testing — in-memory adapter for tests.

Factory functions

Port contracts

Exported from the root and from @netscript/queue/ports.

Errors

Exported from the root and from @netscript/queue/errors.

Validation

Exported from the root and from @netscript/queue/validation.

Adapters

Each provider adapter is published under its own sub-path and implements the MessageQueue contract directly. Most applications use createQueue and never import an adapter explicitly.

The kv-polling adapter additionally re-exports the portable KV store contracts from @netscript/kv: KvStore, WatchableKv, KvEntry, KvKey, AtomicMutation, AtomicCheck, AtomicResult, WatchEvent, and their option interfaces (KvListOptions, KvSetOptions, WatchOptions, WatchPrefixOptions), so KV-Connect consumers can type their backing store without a second import.

Testing

Exported from @netscript/queue/testing.

Sub-path exports

The following entrypoints are published alongside the root export. Their symbols are documented in the sections above.

Durable DLQ Inspection and Reprocessing

Terminal failures are captured in the Dead-Letter Queue (DLQ). The following example shows how to inspect dead-lettered entries, count DLQ depth, and programmatically reprocess them using DeadLetterStorePort.

import { assertEquals } from "@std/assert";
import { MemoryQueueAdapter, MemoryDeadLetterStore } from "@netscript/queue/testing";
import type { DeadLetterRecord } from "@netscript/queue";

Deno.test("durable DLQ inspection and reprocessing worked example", async () => {
  const dlqStore = new MemoryDeadLetterStore<string>();
  const queue = new MemoryQueueAdapter<string>({ deadLetterStore: dlqStore });

  // 1. Enqueue a poison message
  await queue.enqueue("poison-message-payload");

  // 2. Consume and nack without requeueing (sending to DLQ)
  const controller = new AbortController();
  const listenPromise = queue.listen(
    async (message, context) => {
      if (message.startsWith("poison")) {
        await context.nack({
          requeue: false,
          reason: "validation_failed",
          errorCode: "POISON_DETECTED",
          errorMessage: "Invalid format: poison payload detected",
        });
        controller.abort();
      } else {
        await context.ack();
      }
    },
    { signal: controller.signal },
  );

  await listenPromise;

  // 3. Inspect the DLQ store
  const depth = await dlqStore.depth();
  assertEquals(depth, 1);

  const failures = await dlqStore.list({ limit: 10 });
  assertEquals(failures.length, 1);
  const failureRecord: DeadLetterRecord<string> = failures[0];
  assertEquals(failureRecord.payload, "poison-message-payload");
  assertEquals(failureRecord.reason, "validation_failed");

  // 4. Reprocess the DLQ (correcting the message payload and enqueuing it again)
  const processedPayloads: string[] = [];
  const reprocessController = new AbortController();

  const reprocessedCount = await dlqStore.reprocess(async (record) => {
    // Correct the payload and re-enqueue
    const corrected = record.payload.replace("poison-", "clean-");
    await queue.enqueue(corrected);
  });

  assertEquals(reprocessedCount, 1);
  assertEquals(await dlqStore.depth(), 0); // Reprocessed records are removed from DLQ

  // 5. Consume the clean reprocessed message
  const cleanListenPromise = queue.listen(
    async (message, context) => {
      processedPayloads.push(message);
      await context.ack();
      reprocessController.abort();
    },
    { signal: reprocessController.signal },
  );

  await cleanListenPromise;
  assertEquals(processedPayloads, ["clean-message-payload"]);
});

See it live

• How-to: Queue / KV / cron — enqueue, consume, and schedule against these symbols end to end.
• How-to: Choose a queue provider — pick and pin a backend from the QueueProvider enum above.
• Concept: KV, queues & cron — why a queue is provider-agnostic and how Aspire wires the four backends.

Back to the reference overview.