Custom Event Bus

The default event bus runs in-process - fast and zero-configuration for single-instance apps. Multi-instance deployments require events to cross process boundaries.

When you need this

Use a custom event bus when:

• You are running multiple instances of your Slingshot app behind a load balancer
• You need SSE clients on one instance to receive events emitted on another
• You want durable event delivery with a message queue
• You are building a distributed system that shares events across services

For single-instance apps, the in-process bus is the right choice.

BullMQ event bus

For durable, at-least-once delivery with retries and dead-letter queues:

bun add @lastshotlabs/slingshot-bullmq bullmq ioredis

app.config.ts

import { defineApp } from '@lastshotlabs/slingshot';
import { createBullMQAdapter } from '@lastshotlabs/slingshot-bullmq';

export default defineApp({
  eventBus: createBullMQAdapter({
    connection: { host: process.env.REDIS_HOST ?? '127.0.0.1', port: 6379 },
    prefix: 'myapp:events',
    attempts: 5,
  }),
  // ...
});

BullMQ-backed events survive process restarts and can be retried on failure. Use this for anything where losing an event is unacceptable, such as payment webhooks, notification delivery, or audit logging.

Kafka event bus and connectors

Use @lastshotlabs/slingshot-kafka when Slingshot events need to cross process boundaries through Kafka, or when you need to bridge the internal event bus to external Kafka topics.

bun add @lastshotlabs/slingshot-kafka kafkajs

app.config.ts

import { defineApp } from '@lastshotlabs/slingshot';
import { createKafkaAdapter, createKafkaConnectors } from '@lastshotlabs/slingshot-kafka';

const eventBus = createKafkaAdapter({
  brokers: ['127.0.0.1:9092'],
  clientId: 'my-app',
});

const kafkaConnectors = createKafkaConnectors({
  brokers: ['127.0.0.1:9092'],
  inbound: [
    {
      topic: 'orders.created',
      groupId: 'billing-sync',
      handler: async (payload, metadata) => {
        console.log(metadata.topic, payload);
      },
    },
  ],
  outbound: [
    {
      event: 'billing:invoice.created',
      topic: 'billing.invoice.created',
    },
  ],
});

export default defineApp({
  eventBus,
  kafkaConnectors,
  // ...
});

Kafka is the right fit when another system already standardizes on Kafka, when you need durable replayable topics, or when platform teams own the broker and want applications to integrate at the transport boundary instead of inventing a custom bus.

Redis event bus

For a lightweight pub/sub bridge between instances, implement SlingshotEventBus against ioredis:

bun add ioredis

src/eventBus.ts

// @skip-typecheck
import { Redis } from 'ioredis';
import type {
  EventEnvelope,
  SlingshotEventBus,
  SlingshotEventMap,
  SubscriptionOpts,
} from '@lastshotlabs/slingshot-core';

type RedisPayloadHandler = (payload: never) => void | Promise<void>;
type RedisEnvelopeHandler = (envelope: never) => void | Promise<void>;

class RedisEventBus implements SlingshotEventBus {
  private readonly pub: Redis;
  private readonly sub: Redis;
  private readonly handlers = new Map<string, Set<RedisPayloadHandler>>();

  constructor(
    url: string,
    private readonly channel: string,
  ) {
    this.pub = new Redis(url);
    this.sub = new Redis(url);

    this.sub.subscribe(channel);
    this.sub.on('message', (_ch, msg) => {
      const { key, payload } = JSON.parse(msg) as { key: string; payload: unknown };
      this.handlers.get(key)?.forEach(handler => handler(payload as never));
    });
  }

  emit<K extends keyof SlingshotEventMap>(event: K, payload: SlingshotEventMap[K]): void;
  emit(event: string, payload: unknown): void;
  emit(key: string, payload: unknown): void {
    this.pub.publish(this.channel, JSON.stringify({ key, payload }));
  }

  on<K extends keyof SlingshotEventMap>(
    event: K,
    listener: (payload: SlingshotEventMap[K]) => void | Promise<void>,
    opts?: SubscriptionOpts,
  ): void;
  on(event: string, listener: RedisPayloadHandler, opts?: SubscriptionOpts): void;
  on(key: string, handler: RedisPayloadHandler): void {
    if (!this.handlers.has(key)) this.handlers.set(key, new Set());
    this.handlers.get(key)?.add(handler);
  }

  onEnvelope<K extends keyof SlingshotEventMap>(
    event: K,
    listener: (envelope: EventEnvelope<K>) => void | Promise<void>,
    opts?: SubscriptionOpts,
  ): void;
  onEnvelope(event: string, listener: RedisEnvelopeHandler, opts?: SubscriptionOpts): void;
  onEnvelope(): void {}

  off<K extends keyof SlingshotEventMap>(
    event: K,
    listener: (payload: SlingshotEventMap[K]) => void,
  ): void;
  off(event: string, listener: (payload: unknown) => void): void;
  off(key: string, handler: RedisPayloadHandler): void {
    this.handlers.get(key)?.delete(handler);
  }

  offEnvelope<K extends keyof SlingshotEventMap>(
    event: K,
    listener: (envelope: EventEnvelope<K>) => void,
  ): void;
  offEnvelope(event: string, listener: (envelope: EventEnvelope) => void): void;
  offEnvelope(): void {}

  async shutdown(): Promise<void> {
    await Promise.all([this.pub.quit(), this.sub.quit()]);
  }
}

export function createRedisEventBus(opts: { url: string; channel: string }) {
  return new RedisEventBus(opts.url, opts.channel);
}

Then wire it up in app.config.ts:

app.config.ts

// @skip-typecheck
import { defineApp } from '@lastshotlabs/slingshot';
import { createRedisEventBus } from './src/eventBus';

export default defineApp({
  eventBus: createRedisEventBus({
    url: process.env.REDIS_URL ?? 'redis://127.0.0.1:6379',
    channel: 'slingshot:events',
  }),
  // ...
});