Custom PubSub & SSE

The module includes a default Pub/Sub system out of the box. It uses a Nitro hook (zenstack:realtime) to pass mutation events from the database hooks directly to the Server-Sent Events (SSE) handler.

While this works well for a single-server Node.js runtime, it does not scale across multiple server instances or edge workers. If you're hosting on platforms that scale out or serverless architectures where memory is not shared, you will need a distributed Pub/Sub mechanism.

Providing a Custom Pub/Sub

You can replace the built-in system by calling provideZenstackPubsub within a Nitro plugin. You must supply a publish, subscribe, and unsubscribe method that handles ServerRealtimeData.

The following example uses Upstash for serverless PubSub support.

// server/plugins/realtime.ts
import { Redis } from '@upstash/redis'
import { Realtime } from '@upstash/realtime'
import { z } from 'zod'
import type { ServerRealtimeData } from '#build/types/nuxt-zenstack'

export default defineNitroPlugin(() => {
    const url = '_UPSTASH_REDIS_REST_URL_'
    const token = '_UPSTASH_REDIS_REST_TOKEN_'
    const channelName = '_UPSTASH_REDIS_REST_CHANNEL_'

    const schema = {
        data: z.object({
            model: z.string(),
            action: z.string(),
            ids: z.array(z.string().or(z.number()))
        })
    }

    const redis = new Redis({ url, token })
    const realtime = new Realtime({ schema: schema, redis: redis })
    const channel = realtime.channel(channelName)

    provideZenstackPubsub({
        publish: (args) => channel.emit('data', args),
        subscribe: async (cb) => {
            await channel.subscribe({
                events: ['data'],
                onData: ({ data }) => {
                    cb(data as ServerRealtimeData)
                }
            })
        },
        unsubscribe: () => {
            channel.unsubscribe()
        }
    })
})