Realtime

The broadcast channel delivers a notification to a recipient's live connection, the moment it happens: a "poke" on top of the durable database channel, not a replacement for it. This page covers the gateway that holds those connections.

Why a separate gateway

Sending notifications happens from any process, often the worker, in a listener. Holding client connections is a different job with a different scaling shape: many long-lived streams, one per connected user, that must survive instances coming and going freely. The framework keeps these apart: notify/broadcast is a tiny publisher any process imports (no server, no held state); realtime is a dedicated binary (cmd/realtime, scaffolded by add realtime) that only holds connections and relays what gets published.

The driver seam is deliberate: nothing outside notify/broadcast and realtime imports centrifuge directly, so a different transport can replace it later without touching application code.

Publishing

import "github.com/gp-system/notify/broadcast"

pub := broadcast.MustNewPublisher(ctx, cfg.Realtime.BroadcastValkey(), cfg.Realtime.Broadcast)
notifier := notify.NewHub(
    notify.NewMailChannel(mailer),
    notifydatabase.NewChannel(store),
    notify.NewBroadcastChannel(pub, cfg.Realtime.Broadcast.ChannelPrefix),
)

Publisher is a publish-only centrifuge.Node sharing the project's Valkey broker; it never accepts a connection, so a worker process depends on nothing but a network client. Publishing is fire-and-forget: a failed publish is returned to the caller, not retried. Durability lives in the database channel, and a client that misses a live push backfills from it on reconnect. Insert-then-publish is the contract notify.Hub.Send already enforces (database channel before broadcast, same delivery ID in both).

The gateway

cmd/realtime holds no business dependencies (no database pool, no outbox), which is exactly why it scales freely. It exposes WebSocket plus an HTTP-streaming/SSE fallback (all through centrifugal/centrifuge, the library behind Centrifugo and Grafana Live) under /realtime/*, authenticates each connection's JWT in the connect frame (no separate /broadcasting/auth endpoint; v1's only private channel is a connection's own subject, so authenticating once at connect is sufficient), and server-side subscribes every connection to its own user:<subject> channel. No client-side subscribe logic needed for the common case.

Shared topic channels are opt-in and explicitly authorized: each connection's JWT resolves to an rbac.Identity, and a topic is only ever subscribable if some Allow/AllowPrefix callback vouches for that identity.

err := realtime.Run(ctx, cfg.Realtime, func(topics *realtime.TopicAuth) error {
    topics.Allow("announcements", func(id *rbac.Identity) bool { return true })
    topics.Allow("admin-alerts", func(id *rbac.Identity) bool { return id.HasRole("admin") })
    return nil
})

A topic with no registered Allow callback is never subscribable.

For a channel family whose full name is only known at runtime, one per row of some dynamic resource (a tenant, a project, a repository), naming each one individually with Allow at startup is not possible: use AllowPrefix instead, whose callback receives the full topic string so it can look up which specific instance is being requested.

topics.AllowPrefix("project:", func(id *rbac.Identity, topic string) bool {
    projectID, _ := strings.CutPrefix(topic, "project:")
    return projectAccess.CanView(id.Subject, projectID)
})

An exact Allow match always takes priority over a matching AllowPrefix; among prefixes, the first one registered that matches wins.

Client-to-server RPC

By default the gateway is server-to-client only: a connection receives publishes but cannot write anything. WithRPC turns on the other direction: a client sends a method name and a raw payload, and the server answers with a registered RPCHandler, seeing the same JWT-resolved identity a TopicAuth callback does.

err := realtime.Run(ctx, cfg.Realtime, register, realtime.WithRPC(func(rpc *realtime.RPCRegistry) error {
    rpc.Handle("chat.send", func(ctx context.Context, id *rbac.Identity, data []byte) ([]byte, error) {
        return chatService.SendViaRPC(ctx, id, data)
    })
    return nil
}))

A method with no registered handler gets the client centrifuge.ErrorMethodNotFound. The handler's returned error maps much like httperr does: an error carrying an errs.Status becomes the matching centrifuge.Error (400/422 → ErrorBadRequest, 401/403 → ErrorPermissionDenied, 404 → a framework-defined "not found" error), and its errs.Public message is what the client sees; a plain error with no errs.Status maps to ErrorInternal, with no internal detail leaked.

The gateway has no database access (see "holds no business dependencies" above), so an RPC handler typically calls an existing HTTP endpoint (where validation, policy, and the transaction already live) rather than duplicating business logic.

Options and the middleware chain

realtime.Run takes options in the same style as server.Run:

The framework chain around the transport endpoints is aligned with the server chassis, built from the same exported gpsystem/httpmw building blocks, outermost first:

1. CORS, streaming preset (httpmw.CORSStreaming): unlike the API preset, credentials are allowed (some streaming setups authenticate with cookies), so the per-origin decision comes from the origin allower described below instead of a wildcard.
2. request id (httpmw.RequestID): every response carries an X-Request-Id header, same semantics as on the API side.
3. otel (httpmw.Otel("realtime")): the gateway's span; /realtime/healthz is filtered out of tracing, and WithoutTelemetry skips the whole entry.
4. Sentry request hub (httpmw.RequestHub): breadcrumb isolation per request.
5. panic recovery (httpmw.Recoverer): a panic in the HTTP layer becomes a plain 500 plus an error log carrying the stack; the gateway has no Problem writer, and streaming clients reconnect on any 5xx anyway.

WithMiddleware extras run inside these, right before the transport endpoints. There is deliberately no access log: the connection endpoints hold long-lived streams, so a per-request record would be misleading.

Scaling: instances come and go freely

Every gateway instance is stateless: it only ever holds the channels its own connections need, and the shared Valkey broker fans a publish out to every instance that needs it, from any process, without a sticky-session requirement.

• Scale up: docker compose up -d --scale realtime=N; a new instance starts with zero state and accumulates subscriptions only as clients land on it.
• Scale down / rolling restart: SIGTERM flips a draining flag (/realtime/healthz and new connect attempts return 503, so the load balancer's health check evicts the instance), node.Shutdown disconnects every held connection with a reconnect-eligible code, closers run, telemetry flushes. The client's own reconnect logic (backoff, built into the JS client) lands it on a surviving instance, which subscribes it fresh; anything published during the gap is recovered from the database channel, not replayed by the gateway.
• Crash: the client side is identical to a graceful restart, just without the drain step.
• Valkey restart: reconnect and resubscribe are centrifuge's job; publishing during the outage is logged and dropped (the same fire-and-forget contract), recovered the same way as a missed publish during a client's own reconnect.

Config

See the add realtime reference for what gets scaffolded (the binary, compose.yml's Traefik-routed service, .env.example) and the add notification page for wiring the broadcast channel into a notification's Via.

Origin checking

CORS_ORIGINS gives all three transports one rule: same-origin connections (the Origin header's host matches the request's Host, the documented production setup) and clients with no Origin header at all (server-to-server calls, health checks) always pass, regardless of the list; every other case is decided by the list (* allows everything, otherwise an exact match or a pattern containing *, e.g. https://*.example.com). Rejecting a foreign origin logs a Warn.

In development it's common for the frontend and the gateway to be served from different origins (say, next dev on the host at http://localhost:3000, with the gateway behind Traefik at http://<app>.localhost): in that case CORS_ORIGINS must include the frontend's origin (or stay * in dev), otherwise the websocket handshake gets a 403.

Client

import { Centrifuge } from 'centrifuge'

const client = new Centrifuge('wss://your-app.example/realtime/connection/websocket', {
    getToken: async () => fetchAccessToken(), // the same JWT the API uses
})
client.on('publication', (ctx) => {
    // ctx.data: { id, name, payload, at }, id ties back to the database channel's row
})
client.connect()

centrifuge-js handles reconnect backoff and token refresh; the server-side subscription to the recipient's own channel means the client needs no explicit subscribe call for notifications addressed to it. On reconnect, fetch unread notifications from your module's own inbox endpoint (built on the database channel) to backfill anything missed while disconnected. The socket is a live poke, the database is the source of truth.

If the gateway registered a method through WithRPC, the client calls it with client.rpc(...):

const reply = await client.rpc('chat.send', { body: 'hi' })
// reply.data: the handler's raw response, JSON.parse is the caller's job

A failed call throws the centrifuge.Error (code, message); see the "Client-to-server RPC" section above for the error mapping.

Related pages

• notify/broadcast: the publish-only side any process imports.
• auth/rbac: the Identity a connection's JWT resolves to, and what Allow/AllowPrefix check against.
• Application lifecycle: the shared app.Run skeleton realtime.Run builds on.