npm package: @polinater/qstash-event-bus
Description: QStash-backed event bus with strict type safety and automatic retries
Latest version: 5.0.1

# QStash Event Bus

A type-safe event bus powered by [Upstash QStash](https://upstash.com/qstash). Zero Redis. One factory.

---

## Installation

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

## 1 · Extract Prisma types (optional but recommended)

```ts
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)

```ts
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

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

await eventBus.emit(EventType.User.Created, {
  metadata: { eventId: crypto.randomUUID() },
  data: { id: 'user_1', email: 'user@example.com' },
});
```

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

```ts
// 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](https://upstash.com/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

```bash
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:

```bash
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](https://console.upstash.com/qstash).

```bash
# .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.

```typescript
// 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.

```typescript
// 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.

```typescript
// 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.

```typescript
// 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**:
```typescript
// 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**:
```typescript
// 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**:
```bash
# 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**:
```typescript
// Clean up event handlers
await eventBus.off(EventType.BusinessCreated);

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

### Debug Mode

Enable detailed logging:

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

### Health Checks

Implement health checks for monitoring:

```typescript
// 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

```typescript
// 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

```typescript
// 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

```typescript
// Configure Redis connection
configureRedisEventBus(config);

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

### Type Utilities

```typescript
// 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](https://github.com/your-org/redis-event-bus).

## License

MIT License - see LICENSE file for details.