QStash Event Bus

A type-safe event bus powered by Upstash QStash. Zero Redis. One factory.

Installation

npm install @polinater/qstash-event-bus @upstash/qstash

1 · Extract Prisma types (optional but recommended)

import { PrismaClient } from '@prisma/client';
import type { ExtractPrismaTypes } from '@polinater/qstash-event-bus';

const prisma = new PrismaClient();

// Produces a mapping of Model → type
export type MyPrismaTypes = ExtractPrismaTypes<typeof prisma>;

2 · Create the event bus (singleton)

import { createEventBus } from '@polinater/qstash-event-bus';

export const eventBus = createEventBus<MyPrismaTypes>({
  // URL where your API route/webhook receives events
  baseSubscriptionUrl: process.env.BASE_SUBSCRIPTION_API_URL!,
});

Required environment variables:

• QSTASH_TOKEN – your QStash publish token
• QSTASH_CURRENT_SIGNING_KEY – current webhook signing key
• QSTASH_NEXT_SIGNING_KEY – next signing key (rotation)

3 · Emit events

import { EventType, eventBus } from './path/to/bus';

await eventBus.emit(EventType.User.Created, {
  metadata: { eventId: crypto.randomUUID() },
  data: { id: 'user_1', email: '[email protected]' },
});

4 · Handle QStash webhook (Next.js / Express / etc.)

// pages/api/qstash/[...path].ts  (Next.js example)
import type { NextApiRequest, NextApiResponse } from 'next';
import { createQStashHandler, eventBus } from '@/lib/event-bus';

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  const qstashHandler = createQStashHandler(eventBus);
  // The handler deals with verification & routing
  await qstashHandler(
    { method: req.method, headers: req.headers as any, body: req.rawBody, url: req.url },
    { status: code => ({ json: body => res.status(code).json(body), send: msg => res.status(code).send(msg) }) },
  );
}

createQStashHandler:

1. Verifies the signature via Upstash keys.
2. Converts the request path (/user/created) ➜ User.Created.
3. Routes the event to any handlers registered with eventBus.on.

API

| Function / Type | Description | |-----------------|-------------| | createEventBus<TPrismaTypes>(config) | Returns the singleton QStashEventBus<TPrismaTypes> | | eventBus.emit(event, payload) | Publish an event to QStash | | eventBus.on(event, handler) | Register a handler for an event | | createQStashHandler(bus) | Returns an HTTP handler that validates & routes QStash webhooks | | ExtractPrismaTypes<typeof prisma> | Utility type to map Prisma models | | EventType | Enum of all events |

License

MIT

A type-safe, Prisma-integrated event bus for Node.js and Next.js, powered by Upstash QStash.

This package provides a simple, familiar API (emit, on, off) for building robust, event-driven architectures without the need for managing your own queue infrastructure.

What’s New – The QStash Migration

This version marks a complete migration from a self-hosted Redis/BullMQ transport layer to a serverless model using Upstash QStash.

• Serverless Architecture: No more long-running worker processes to manage. Events are published via HTTP and consumed via webhooks.
• Simplified Setup: Configuration is now centered around environment variables and a single base URL for your API endpoints.
• URL-based Events: Event names (e.g., Order.Created) are automatically mapped to webhook URLs (e.g., /api/events/order/created).
• Built-in Retries & Durability: Inherits QStash's at-least-once delivery guarantee with automatic retries and a dead-letter queue.

The core TypeScript API remains consistent, preserving the type-safe developer experience you're used to.

Installation

npm install @polinater/qstash-event-bus @upstash/qstash

Prerequisites

• Node.js 16 or higher
• An Upstash account and QStash credentials
• Prisma Client (optional but recommended for type safety)
• Next.js 13+ (for App Router examples)

Peer Dependencies

The package works with Prisma v5 and v6:

npm install @prisma/client # ^5.0.0 || ^6.0.0

Setup Guide

Step 1: Environment Configuration

Create or update your .env.local file with your QStash credentials. You can find these in the Upstash Console.

# .env.local

# Required: Your QStash bearer token for publishing events.
QSTASH_TOKEN="..."

# Required: Keys for verifying incoming webhooks. Find these in the Upstash console.
QSTASH_CURRENT_SIGNING_KEY="..."
QSTASH_NEXT_SIGNING_KEY="..."

# Required: The base URL of your application's API for receiving events.
BASE_SUBSCRIPTION_API_URL="http://localhost:3000/api/events"

Step 2: Initialize Event Bus

Create a centralized, singleton instance of the event bus.

// src/lib/eventBus.ts
import { createEventBus } from '@polinater/qstash-event-bus';

export const eventBus = createEventBus({
  baseSubscriptionUrl: process.env.BASE_SUBSCRIPTION_API_URL!,
});

Step 3: Define Event Handlers

Define the functions that will run when you receive an event. You can centralize these in a single file.

// src/lib/subscribers.ts
import { eventBus } from './eventBus';
import { EventType } from './types/events'; // Assuming you have this enum defined

export function initializeSubscribers() {
  eventBus.on(EventType.OrderCreated, async (event) => {
    console.log('Order created:', event.data.orderId);
    // Send confirmation email, update analytics, etc.
  });

  eventBus.on(EventType.UserRegistered, async (event) => {
    console.log('New user registered:', event.data.userId);
    // Send welcome email...
  });

  console.log('Event handlers registered.');
}

Step 4: Create a Webhook Endpoint

Create a catch-all API route in your Next.js application to receive, verify, and route all incoming events from QStash.

// src/app/api/events/[...slug]/route.ts
import { createQStashHandler } from '@polinater/qstash-event-bus';
import { eventBus } from '@/lib/eventBus';
import { initializeSubscribers } from '@/lib/subscribers';

// Register your event handlers once
initializeSubscribers();

// Create the handler that verifies and routes incoming requests
const handler = createQStashHandler(eventBus);

export { handler as POST };

This single endpoint will handle all events. For example, an Order.Created event will be sent to /api/events/order/created, and the handler will automatically match it to the correct .on() handler.

Usage Example: Emitting an Event

Once set up, emitting an event from your server-side code (e.g., in another API route) is straightforward.

// src/app/api/orders/route.ts
import { NextRequest, NextResponse } from 'next/server';
import { eventBus } from '@/lib/eventBus';
import { EventType } from '@/lib/types/events';
import { prisma } from '@/lib/prisma';

export async function POST(request: NextRequest) {
  try {
    const data = await request.json();

    // 1. Create the order in your database
    const order = await prisma.order.create({ data });

    // 2. Emit an event
    await eventBus.emit(EventType.OrderCreated, {
      metadata: {
        eventId: crypto.randomUUID(),
        timestamp: new Date().toISOString(),
        version: '1.0',
      },
      data: {
        orderId: order.id,
      },
    });

    return NextResponse.json(order, { status: 201 });
  } catch (error) {
    console.error('Failed to create order:', error);
    return NextResponse.json({ error: 'Server error' }, { status: 500 });
  }
}

3. Persistence: Events are not persisted by default - consider Redis persistence settings
4. Monitoring: Monitor Redis memory usage and connection counts

Troubleshooting

Common Issues

1. Events Not Being Received

Problem: Subscribers not receiving events Solutions:

// Ensure subscribers are initialized before emitting
await initializeSubscribers();
await new Promise(resolve => setTimeout(resolve, 1000)); // Brief delay
await eventBus.emit(EventType.BusinessCreated, payload);

// Check Redis connection
const isHealthy = await eventBus.isHealthy();
console.log('Redis healthy:', isHealthy);

2. Type Errors with Prisma

Problem: TypeScript errors with Prisma types Solutions:

// Ensure Prisma client is generated
// Run: npx prisma generate

// Use type assertion if needed
import type { Business } from '@prisma/client';
const business = event.data.data as Business;

// Or regenerate types
import type { ExtractPrismaTypes } from '@polinater/redis-event-bus';
import { PrismaClient } from '@prisma/client';
type MyTypes = ExtractPrismaTypes<PrismaClient>;

3. Redis Connection Issues

Problem: Cannot connect to Redis Solutions:

# Check Redis is running
redis-cli ping

# Check environment variables
echo $REDIS_EVENTS_URL

# Test connection manually
redis-cli -u $REDIS_EVENTS_URL ping

4. Memory Leaks

Problem: Memory usage growing over time Solutions:

// Clean up event handlers
await eventBus.off(EventType.BusinessCreated);

// Implement connection cleanup
process.on('exit', async () => {
  await eventBus.disconnect();
});

Debug Mode

Enable detailed logging:

configureRedisEventBus({
  enableLogging: true,
  options: {
    showFriendlyErrorStack: true,
    maxmemoryPolicy: 'allkeys-lru'
  }
});

Health Checks

Implement health checks for monitoring:

// app/api/health/redis/route.ts
import { eventBus } from '@/lib/eventBus';

export async function GET() {
  try {
    const isHealthy = await eventBus.isHealthy();
    
    if (isHealthy) {
      return Response.json({ status: 'healthy', redis: 'connected' });
    } else {
      return Response.json(
        { status: 'unhealthy', redis: 'disconnected' }, 
        { status: 503 }
      );
    }
  } catch (error) {
    return Response.json(
      { status: 'error', message: error.message }, 
      { status: 500 }
    );
  }
}

API Reference

Factory Functions

// Create with default types
const eventBus = createEventBus();

// Create with Prisma types
const eventBus = createEventBusWithPrisma<MyPrismaTypes>();

// Create from Prisma client (recommended)
const eventBus = createEventBusFromClient(prismaClient);

Event Bus Methods

// Subscribe to events
await eventBus.on(eventType, handler);

// Unsubscribe from events
await eventBus.off(eventType, handler?);

// Emit events
await eventBus.emit(eventType, payload);

// Health checks
const healthy = await eventBus.isHealthy();
const connected = eventBus.getConnectionStatus();

// Cleanup
await eventBus.disconnect();

Configuration

// Configure Redis connection
configureRedisEventBus(config);

// Configuration interface
interface RedisEventBusConfig {
  url?: string;
  options?: RedisOptions;
  enableLogging?: boolean;
}

Type Utilities

// Extract types from Prisma client
type MyTypes = ExtractPrismaTypes<typeof prismaClient>;

// Infer types from instance
type MyTypes = InferPrismaTypes<typeof prismaClient>;

Support

For issues, questions, or contributions, please visit the GitHub repository.

License

MIT License - see LICENSE file for details.