Broadcasting Guide

Guren provides a broadcasting system for real-time event broadcasting to connected clients. This is useful for building features like live notifications, chat applications, and real-time dashboards.

Core Concepts

• BroadcastManager – Central hub for managing channels, drivers, and SSE clients.
• Channel – A named conduit for broadcasting events. Channels can be public, private, or presence.
• BroadcastDriver – Backend for event distribution (Memory or Redis).
• SSE (Server-Sent Events) – Built-in support for pushing events to browser clients.
• WebSockets – A socket endpoint on the same channels, plus lifecycle helpers for custom socket routes.

Channel Types

• Public Channels – Anyone can subscribe.
• Private Channels – Require user authentication to subscribe.
• Presence Channels – Track who is subscribed (e.g., for "who's online" features).

Basic Usage

Setup

import { BroadcastManager } from '@guren/core'

const broadcast = new BroadcastManager({
  default: 'memory',
  drivers: {
    memory: () => new MemoryDriver(),
  },
})

// Broadcast an event
await broadcast.broadcast('notifications', 'NewMessage', {
  content: 'Hello world!',
})

Container Integration

The broadcasting subsystem is registered as a singleton via a ServiceProvider. You can resolve it from the container:

// Access via app.container or this.container in providers

const broadcast = container.make('broadcast') // BroadcastManager
await broadcast.broadcast('notifications', 'NewMessage', { content: 'Hello!' })

Using Channel Helpers

// Public channel
await broadcast.toChannel('notifications').broadcast('NewMessage', data)

// Private channel (auto-prefixed with 'private-')
await broadcast.toPrivate('orders.123').broadcast('OrderUpdated', {
  status: 'shipped',
})

// Presence channel (auto-prefixed with 'presence-')
await broadcast.toPresence('chat.general').broadcast('UserJoined', {
  user: 'John',
})

Subscribing to Channels

const channel = broadcast.toChannel('notifications')

// Subscribe to events
const unsubscribe = channel.subscribe((event, data) => {
  console.log(`Event: ${event}`, data)
})

// Later, unsubscribe
unsubscribe()

Channel Authorization

Public Channels

broadcast.channel('notifications', () => true)
broadcast.channel('public.*', () => true) // Wildcard pattern

Private Channels

Private channels require authentication:

broadcast.privateChannel('orders.{orderId}', async (channel, user) => {
  // Extract orderId from channel name
  const orderId = channel.replace('private-orders.', '')

  // Check if user owns this order
  const order = await Order.find(orderId)
  return order?.userId === user.id
})

// User-specific channels
broadcast.privateChannel('user.{userId}', (channel, user) => {
  const userId = channel.replace('private-user.', '')
  return String(user.id) === userId
})

Presence Channels

Presence channels return member info when authorized:

broadcast.presenceChannel('chat.{roomId}', async (channel, user) => {
  if (!user) return null // Not authorized

  // Return presence member info
  return {
    id: user.id,
    info: {
      name: user.name,
      avatar: user.avatar,
    },
  }
})

Pattern Matching

Channel patterns support:

• {param} – Match any non-dot segment
• * – Match any single segment
• ** – Match multiple segments

broadcast.channel('posts.*', () => true)           // posts.123, posts.456
broadcast.channel('users.{id}.posts', authorizer)  // users.1.posts
broadcast.channel('admin.**', isAdmin)             // admin.users, admin.settings.email

Server-Sent Events (SSE)

SSE Endpoint

import { AUTH_CONTEXT_KEY, Router } from '@guren/core'
import type { AuthContext } from '@guren/core'
import type { Context } from 'hono'

const currentUser = async (ctx: Context) => {
  const auth = ctx.get(AUTH_CONTEXT_KEY) as AuthContext | undefined
  return (await auth?.user()) ?? null
}

export function registerBroadcastRoutes(router: Router): void {
  router.get('/broadcasting/events', broadcast.sseMiddleware({
    pingInterval: 30000,
    retry: 3000,
    // Resolve the connecting user so channels requested up front via
    // ?channels= can be authorized when the stream opens
    getUser: (ctx) => currentUser(ctx as Context),
  }))

  router.post('/broadcasting/auth', broadcast.authMiddleware({
    getUser: (ctx) => currentUser(ctx as Context),
  }))
}

getUser receives the request context typed as unknown and may return a promise. currentUser() reads the signed-in user from the auth context, and the examples below reuse it.

The SSE endpoint accepts a ?channels= query parameter listing channels to subscribe before the stream starts. Each requested channel is authorized against the user returned by getUser, so a plain EventSource works for public channels with zero extra calls. Private and presence channels are subscribed later through /broadcasting/auth (see Authorizing Channels (Client)).

WebSocket Foundation

broadcast.webSocketMiddleware() is the WebSocket counterpart of the SSE endpoint. Mount it on a GET route and the request is upgraded to a socket that carries the same broadcast events, through the same channel authorizers and drivers:

import { Router } from '@guren/core'

export function registerBroadcastRoutes(router: Router): void {
  router.get('/broadcasting/socket', broadcast.webSocketMiddleware({
    // The user of the upgrade request authorizes every channel the socket
    // subscribes to, for as long as it stays open
    getUser: (ctx) => currentUser(ctx as Context),
  }))
}

The upgrade needs a Bun server, which app.listen() provides: it hands Bun.serve the WebSocket handler that upgradeWebSocket from hono/bun relies on. On a runtime that cannot upgrade (app.fetch() on Node, Workers, Lambda) the route answers 501.

Every frame is JSON. The server sends { event, data }, starting with a connected event that carries the clientId and the channels subscribed from ?channels=. The client subscribes and unsubscribes by sending { action, channel }. Each message is authorized like a POST /broadcasting/auth request and answered with a subscription event:

const socket = new WebSocket(
  `${location.origin.replace(/^http/, 'ws')}/broadcasting/socket?channels=announcements`
)

socket.addEventListener('open', () => {
  socket.send(JSON.stringify({ action: 'subscribe', channel: 'private-orders.123' }))
})

socket.addEventListener('message', (e) => {
  const { event, data } = JSON.parse(e.data)
  if (event === 'subscription') {
    // e.g. { channel: 'private-orders.123', authorized: true, subscribed: true }
    console.log('Subscription:', data)
  } else if (event === 'OrderUpdated') {
    console.log('Order updated:', data)
  }
})

// Later
socket.send(JSON.stringify({ action: 'unsubscribe', channel: 'private-orders.123' }))

Messages are handled in the order they arrive. A refused channel is answered with authorized: false and delivers nothing. A socket with more than 32 messages waiting for an answer is closed with code 1008, and a message over 4 KB is ignored. The clientId from connected also works with POST /broadcasting/auth.

The socket keeps the user it opened with, so signing out does not close it. Remove that user's clients when they sign out or lose access:

for (const client of broadcast.getWebSocketClients()) {
  if (client.userId === user.id) broadcast.removeWebSocketClient(client.id)
}

Origin check

CORS does not apply to a WebSocket handshake, and the browser sends the app's cookies with one opened from any site. A page elsewhere could therefore open a socket as the signed-in user (Cross-Site WebSocket Hijacking). The route refuses with 403 a handshake whose Origin names a host other than the request's own. When a proxy in front of the app rewrites Host, list the public origin:

broadcast.webSocketMiddleware({
  getUser: (ctx) => currentUser(ctx as Context),
  allowedOrigins: ['https://app.example.com'],
})

A handshake with no Origin comes from outside a browser and carries none of the user's cookies. It passes the check and still has to authorize its channels like any other socket. The check compares hosts and ignores the scheme, because TLS usually ends at the proxy, so a plain-HTTP page on the same host passes; HSTS is what keeps pages on HTTPS. An allowedOrigins entry matches its scheme exactly.

Custom socket routes

The lower-level helpers remain for a route with its own protocol. subscribeWebSocketClient() does not authorize, like subscribeClient() for SSE: pass it a channel the server chose, and run a channel the client named through broadcast.authorize() first. Registering the client with the user's id lets POST /broadcasting/auth attach channels to it, for that user only. In the socket's open handler:

const clientId = broadcast.registerWebSocketClient({
  userId: user.id,
  send: (event, data) => ws.send(JSON.stringify({ event, data })),
  close: () => ws.close(),
})

if (await broadcast.authorize(channel, user)) {
  broadcast.subscribeWebSocketClient(clientId, channel)
}

// When the socket closes
broadcast.removeWebSocketClient(clientId)

Put createWebSocketOriginGuard() in front of such a route for the same Origin check; it takes the same allowedOrigins:

import { createWebSocketOriginGuard } from '@guren/core'

router.get('/socket', socketHandler, createWebSocketOriginGuard())

broadcast.disconnectAll() closes WebSocket clients as well as SSE streams.

Typed channel codegen

guren codegen now generates .guren/channels.gen.ts from your server-side broadcast usage.

// app/Providers/BroadcastProvider.ts
broadcast.channel('announcements', () => true)
// A private channel needs a real check — `() => true` would let any
// caller subscribe to another user's channel.
broadcast.privateChannel('posts.{id}', async (channel, user) => {
  const post = await Post.find(channel.replace('private-posts.', ''))
  return post?.authorId === user?.id
})
broadcast.broadcast('announcements', 'NewPost', { id: 1 })

Generated artifacts include:

• ChannelName – channel name union with pattern-aware template literal types.
• ChannelEvents – channel-to-event map with inferred payload types for literal/object/array broadcast payloads.
• channelEventManifest – runtime manifest for discovered channels/events.

import type { ChannelEvents } from '@/.guren/channels.gen'
import { createUseChannel } from '@guren/inertia-client'

const useChannel = createUseChannel<ChannelEvents>()
const feed = useChannel('announcements')
const off = feed.on('NewPost', (payload) => {
  console.log(payload)
})

This gives you typed channel/event names and inferred payload shapes in the frontend.

Each useChannel(name) call opens its own EventSource on endpoint?channels=name (default endpoint /broadcasting/events; an endpoint that already has a query string gets &channels=). The channel argument is what the server subscribes, not just a type. The stream is authorized and subscribed up front exactly like the ?channels= example below, so a private or presence channel works through the same call when the SSE route resolves the user with getUser. A channel the server refuses is left out of the connected event's channels list and delivers nothing. One stream per channel is deliberate: events are dispatched by event name, so keeping each channel on its own stream is what lets feed.on('NewPost', …) mean "NewPost on announcements". channelStreamUrl(endpoint, channel) is exported for building that URL by hand.

End-to-end typed realtime flow

Use the generated ChannelEvents on the server too, so emit-side payloads are checked at compile time.

import type { ChannelEvents } from '@/.guren/channels.gen'
import { createTypedBroadcaster } from '@guren/core'

const typed = createTypedBroadcaster<ChannelEvents>(broadcast)

await typed.broadcast('announcements', 'NewPost', { id: 1 }) // typed payload
await typed.toChannel('announcements').broadcast('NewPost', { id: 2 })

Client-Side Integration

Pass public channels in the ?channels= query parameter to subscribe them as soon as the stream opens. Right after connecting, the server sends a connected event carrying your clientId and the list of channels that were authorized and subscribed. Capture the clientId, because you need it to subscribe to private and presence channels later.

// Connect to SSE and subscribe public channels up front
const eventSource = new EventSource(
  '/broadcasting/events?channels=notifications,announcements'
)

// The server sends a `connected` event first — capture the clientId
let clientId: string | null = null

eventSource.addEventListener('connected', (e) => {
  const data = JSON.parse(e.data)
  clientId = data.clientId
  console.log('Subscribed channels:', data.channels)
})

// Messages are dispatched by EVENT name, not channel name
eventSource.addEventListener('NewMessage', (e) => {
  const data = JSON.parse(e.data)
  console.log('New message:', data)
})

// Listen for ping
eventSource.addEventListener('ping', () => {
  console.log('Keepalive ping')
})

eventSource.onerror = (error) => {
  console.error('Connection error', error)
}

Authorizing Channels (Client)

Private and presence channels are subscribed through POST /broadcasting/auth. A single request with { clientId, channel } both authorizes the channel for the current user and subscribes your SSE connection (or WebSocket) to it. The response reports both results per channel:

async function subscribeToPrivateChannel(channel: string) {
  if (!clientId) {
    throw new Error('Not connected yet — wait for the `connected` event')
  }

  // Authorize AND subscribe in one call
  const response = await fetch('/broadcasting/auth', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ clientId, channel }),
    credentials: 'include',
  })

  const result = await response.json()
  // e.g. { 'private-orders.123': { authorized: true, subscribed: true } }
  return result[channel]?.authorized && result[channel]?.subscribed
}

if (await subscribeToPrivateChannel('private-orders.123')) {
  // Events arrive on the same EventSource, dispatched by event name
  eventSource.addEventListener('OrderUpdated', (e) => {
    const data = JSON.parse(e.data)
    console.log('Order updated:', data)
  })
}

Configuration

Redis Driver

For production and multi-server support:

import { BroadcastManager, RedisDriver } from '@guren/core'
import { createRedisClient } from '@guren/core'

const redis = createRedisClient({ url: process.env.REDIS_URL })

const broadcast = new BroadcastManager({
  default: 'redis',
  drivers: {
    redis: () => new RedisDriver(redis),
    memory: () => new MemoryDriver(), // Fallback for testing
  },
})

Multiple Drivers

const broadcast = new BroadcastManager({
  default: 'redis',
  drivers: {
    redis: () => new RedisDriver(redis),
    memory: () => new MemoryDriver(),
  },
})

// Use specific driver
const driver = broadcast.driver('memory')
await driver.publish('test-channel', 'TestEvent', data)

Broadcasting from Events

Integrate broadcasting with the event system:

import { Event } from '@guren/core'

export class OrderShipped extends Event {
  constructor(
    public readonly orderId: string,
    public readonly trackingNumber: string
  ) {
    super()
  }

  // Implement BroadcastableEvent interface
  broadcastOn(): string[] {
    return [`private-orders.${this.orderId}`]
  }

  broadcastAs(): string {
    return 'OrderShipped'
  }

  broadcastWith(): Record<string, unknown> {
    return {
      orderId: this.orderId,
      trackingNumber: this.trackingNumber,
    }
  }
}

// Usage
const event = new OrderShipped('123', 'ABC456')

for (const channel of event.broadcastOn()) {
  await broadcast.broadcast(
    channel,
    event.broadcastAs?.() ?? event.eventName,
    event.broadcastWith?.() ?? {}
  )
}

Presence Channel Members

Track who's in a presence channel:

import { PresenceChannel } from '@guren/core'

const channel = broadcast.toPresence('chat.general')

// Get current members (requires PresenceBroadcastDriver)
const driver = broadcast.driver() as PresenceBroadcastDriver
const members = driver.getMembers('presence-chat.general')

// Broadcast member joined
await channel.broadcast('MemberJoined', {
  member: { id: user.id, name: user.name },
})

// Broadcast member left
await channel.broadcast('MemberLeft', {
  memberId: user.id,
})

Testing

Using container.fake()

Swap the broadcast manager in tests:

// Access via app.container or this.container in providers
import { BroadcastManager, MemoryDriver } from '@guren/core'

test('broadcasts order update', async () => {
  const fakeBroadcast = new BroadcastManager({
    default: 'memory',
    drivers: { memory: () => new MemoryDriver() },
  })

  using _ = container.fake('broadcast', fakeBroadcast)

  // Code under test that broadcasts events will use fakeBroadcast
})

Manual Testing

import { describe, test, expect, beforeEach, mock } from 'bun:test'
import { BroadcastManager, MemoryDriver } from '@guren/core'

describe('Broadcasting', () => {
  let broadcast: BroadcastManager

  beforeEach(() => {
    broadcast = new BroadcastManager({
      default: 'memory',
      drivers: {
        memory: () => new MemoryDriver(),
      },
    })
  })

  test('broadcasts to channel', async () => {
    const received: unknown[] = []

    broadcast.driver().subscribe('test', (event) => {
      received.push(event)
    })

    await broadcast.broadcast('test', 'TestEvent', { value: 1 })

    expect(received).toHaveLength(1)
    expect(received[0]).toMatchObject({
      channel: 'test',
      event: 'TestEvent',
      data: { value: 1 },
    })
  })

  test('authorizes private channel', async () => {
    broadcast.privateChannel('orders.{orderId}', (channel, user) => {
      return user?.id === '123'
    })

    const result = await broadcast.authorize('private-orders.456', { id: '123' })
    expect(result).toBe(true)

    const denied = await broadcast.authorize('private-orders.456', { id: '999' })
    expect(denied).toBe(false)
  })

  test('presence channel returns member info', async () => {
    broadcast.presenceChannel('chat.{roomId}', (channel, user) => {
      if (!user) return null
      return { id: user.id, info: { name: user.name } }
    })

    const result = await broadcast.authorize('presence-chat.1', {
      id: '123',
      name: 'John',
    })

    expect(result).toMatchObject({
      id: '123',
      info: { name: 'John' },
    })
  })
})

Best Practices

1. Use Redis in production: Memory driver doesn't work across multiple servers.

2. Authorize sensitive channels: Always protect private and presence channels with proper authorization.

3. Keep payloads small: Broadcast only essential data to reduce bandwidth.

4. Handle disconnections: Implement reconnection logic on the client side.

5. Use presence channels for real-time features: Track online users, typing indicators, etc.

6. Consider message ordering: Events may arrive out of order; include timestamps if order matters.

7. Clean up subscriptions: Always unsubscribe when components unmount or users leave.

8. Test authorization logic: Write tests for channel authorizers to prevent security issues.