@glidemq/hapi

REST API and real-time SSE for glide-mq job queues, as a Hapi.js plugin. Two registrations give you the full queue HTTP surface with Joi validation and Hapi lifecycle hooks.

Turns a Hapi v21 server into a queue management gateway. Built for teams that run Hapi in production and need to expose queue operations to dashboards, CLI tools, or other services.

Package Links

• npm: @glidemq/hapi
• GitHub: github.com/avifenesh/glidemq-hapi

Why @glidemq/hapi

• Use this when you need HTTP endpoints to manage glide-mq queues without writing route handlers yourself.
• Use this when you want live SSE streams of job events for dashboards or monitoring.
• Use this when serverless functions only need to enqueue jobs through lightweight Producer endpoints.
• Use this when you need to test queue logic in CI without a running Valkey instance.

Install

npm install @glidemq/hapi glide-mq @hapi/hapi

This guide uses @glidemq/hapi 0.4.2 with glide-mq 0.17.0.

Quick Start

import Hapi from "@hapi/hapi";
import { glideMQPlugin } from "@glidemq/hapi";

const server = Hapi.server({ port: 3000 });

await server.register({
  plugin: glideMQPlugin,
  options: {
    connection: { addresses: [{ host: "localhost", port: 6379 }] },
    queues: {
      emails: {
        processor: async (job) => {
          console.log("Sending to", job.data.to);
          return { sent: true };
        },
      },
    },
    routes: true,
  },
});

await server.start();

The server now accepts POST /emails/jobs to enqueue jobs and GET /emails/events for live SSE. See the full endpoint table below.

How It Works

glideMQPlugin creates a QueueRegistry, decorates server.glidemq so every route handler can access it, eagerly initializes configured producers, and registers an onPostStop hook that closes all queues, workers, and producers on shutdown. Set routes: true to mount the HTTP API or pass a routes object to restrict queue and producer names. Queue and worker instances are created lazily on first request; producers are created eagerly so connection errors surface at startup.

Endpoints

Method	Route	Description
POST	/{name}/jobs	Add a job to a queue
POST	/{name}/jobs/wait	Add a job and wait for result
GET	/{name}/jobs	List jobs by state
GET	/{name}/jobs/{id}	Get a single job
POST	/{name}/jobs/{id}/priority	Change job priority
POST	/{name}/jobs/{id}/delay	Change job delay
POST	/{name}/jobs/{id}/promote	Promote a delayed job
GET	/{name}/counts	Get job counts by state
GET	/{name}/metrics	Get completed or failed metrics
POST	/{name}/pause	Pause a queue
POST	/{name}/resume	Resume a paused queue
POST	/{name}/drain	Drain all waiting jobs
POST	/{name}/retry	Retry failed jobs
DELETE	/{name}/clean	Clean old completed or failed jobs
GET	/{name}/workers	List active workers
POST	/{name}/produce	Add a job via Producer
GET	/{name}/schedulers	List all schedulers
GET	/{name}/schedulers/{schedulerName}	Get a single scheduler
PUT	/{name}/schedulers/{schedulerName}	Upsert a scheduler
DELETE	/{name}/schedulers/{schedulerName}	Remove a scheduler
GET	/{name}/events	SSE event stream
POST	/flows	Create a tree flow or DAG over HTTP with { flow, budget? } or { dag }
GET	/flows/{id}	Inspect a flow snapshot with nodes, roots, counts, usage, and budget
GET	/flows/{id}/tree	Inspect the nested tree view for a submitted tree flow or DAG
DELETE	/flows/{id}	Revoke or flag remaining jobs in a flow and delete the HTTP flow record
GET	/{name}/flows/{id}/usage	Aggregated token/cost usage across a flow
GET	/{name}/flows/{id}/budget	Budget state (limits, spent, exceeded) for a flow
GET	/{name}/jobs/{id}/stream	SSE stream of real-time chunks from a streaming job
GET	/usage/summary	Rolling token/cost summary across one or more queues
POST	/broadcast/{name}	Publish a broadcast message with a subject and payload
GET	/broadcast/{name}/events	SSE stream for a durable broadcast subscription (?subscription= required)

Features

• SSE event streaming -- subscribe to completed, failed, progress, active, waiting, stalled, usage, suspended, budget-exceeded, and heartbeat events on any queue via GET /{name}/events. Uses PassThrough streams with shared QueueEvents instances (ref-counted per queue).
• Lightweight producers - configure producers for server-side NAPI runtimes that only need to enqueue jobs. The POST /{name}/produce endpoint returns a job ID without requiring a worker.
• Scheduler CRUD -- create, read, update, and delete repeatable jobs through four endpoints. Supports cron patterns, fixed intervals, and repeatAfterComplete mode.
• Testing without Valkey -- createTestApp from @glidemq/hapi/testing spins up an in-memory server backed by TestQueue and TestWorker. Use server.inject() for assertions with no external dependencies.
• Joi validation -- all request bodies, query parameters, and plugin options are validated with Joi schemas and structured error messages.
• Queue access control - pass queues or producers arrays in GlideMQRoutesOptions to restrict which queues the API exposes. Requests to unlisted queues return 404.
• Route prefix - use Hapi's registration options, { routes: { prefix: '/api/queues' } }, to mount the HTTP surface under a path prefix.
• Automatic cleanup -- the onPostStop lifecycle hook closes workers first (to drain in-progress jobs), then queues and producers, using Promise.allSettled for reliability.
• Broadcast over HTTP -- publish messages and stream them via SSE with durable subscriptions and optional subject filters.
• Flow orchestration over HTTP -- create tree flows or DAGs from any HTTP client, then inspect them as flat snapshots or nested trees.

Configuration

interface GlideMQPluginOptions {
  connection?: ConnectionOptions; // Required unless testing: true
  queues?: Record<string, QueueConfig>;
  producers?: Record<string, ProducerConfig>;
  prefix?: string;       // Key prefix for Valkey keys (default: 'glide')
  testing?: boolean;     // Use TestQueue/TestWorker, no Valkey needed
  serializer?: Serializer;
  routes?: boolean | GlideMQRoutesOptions;
}

interface QueueConfig {
  processor?: (job: Job) => Promise<any>; // Omit for producer-only queues
  concurrency?: number;                   // Default: 1
  workerOpts?: Record<string, unknown>;
}

interface ProducerConfig {
  compression?: "none" | "gzip";
  serializer?: Serializer;
}

interface GlideMQRoutesOptions {
  queues?: string[];     // Restrict API to these queue names
  producers?: string[];  // Restrict produce API to these producer names
}

Testing

import { createTestApp } from "@glidemq/hapi/testing";

const { server } = await createTestApp({
  emails: { processor: async (job) => ({ sent: true }) },
});

const res = await server.inject({
  method: "POST",
  url: "/emails/jobs",
  payload: { name: "welcome", data: { to: "user@test.com" } },
});
expect(res.statusCode).toBe(201);

await server.stop();

Direct Registry Access

server.route({
  method: "GET",
  path: "/pending-count",
  handler: async (request, h) => {
    const { queue } = request.server.glidemq.get("emails");
    const counts = await queue.getJobCounts();
    return h.response({ waiting: counts.waiting });
  },
});

Limitations

• Requires a running Valkey or Redis instance for production use. Testing mode uses in-memory stubs only.
• No built-in authentication or authorization. Add Hapi auth strategies or gateway-level controls separately.
• addAndWait (the POST /{name}/jobs/wait endpoint) is not available in testing mode because TestQueue does not support it.
• POST /flows accepts tree flow payloads with optional budgets and DAG payloads without budgets. HTTP-submitted budgets are currently supported for tree flows only.
• /flows*, /usage/summary, and /broadcast/* require a live connection and are unavailable in testing mode.
• Producers are not supported in testing mode. Use queue-based endpoints for test assertions.

Ecosystem

Package	Description
glide-mq	Core queue library
@glidemq/hono	Hono REST + SSE middleware
@glidemq/fastify	Fastify REST + SSE plugin
@glidemq/nestjs	NestJS module with decorators
@glidemq/dashboard	Web UI for queue monitoring

License

Apache-2.0