Realtime

A broadcast csatorna egy notificationt a recipiens élő kapcsolatára kézbesít, abban a pillanatban, amikor történik: egy "poke" a durable database csatorna tetején, nem annak helyettesítője. Ez az oldal a kapcsolatokat tartó gatewayről szól.

Miért külön gateway

A notificationök küldése bármely processből történhet, gyakran a workerből, egy listenerből. A kliens-kapcsolatok tartása más feladat, más skálázási alakkal: sok, hosszú élettartamú stream, egy csatlakozott userenként, aminek túl kell élnie a példányok szabad jövés-menését. A framework ezeket szétválasztja: a notify/broadcast egy pici publisher, amit bármely process importálhat (szerver nélkül, állapot nélkül); a realtime egy dedikált binary (cmd/realtime, az add realtime scaffoldolja), ami csak kapcsolatokat tart és relay-eli, amit publikálnak.

A driver-seam szándékos: a notify/broadcast-on és a realtime-on kívül semmi nem importálja közvetlenül a centrifuge-ot, így egy másik transport később lecserélheti anélkül, hogy az alkalmazáskódhoz nyúlna.

Publikálás

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),
)

A Publisher egy publish-only centrifuge.Node, ami osztozik a projekt Valkey brokerén; sosem fogad kapcsolatot, így egy worker-process semmi mástól nem függ, csak egy hálózati klienstől. A publikálás fire-and-forget: egy sikertelen publish visszaadódik a hívónak, nincs újrapróbálva: a durability a database csatornából jön, egy kliens, aki lemarad egy élő push-ról, reconnectkor onnan backfillel. Az insert-majd-publish kontraktust maga a notify.Hub.Send már kikényszeríti (database csatorna a broadcast előtt, ugyanaz a delivery ID mindkettőben).

A gateway

A cmd/realtime nem hordoz üzleti függőséget (nincs adatbázis-pool, nincs outbox), pont ezért skálázódik szabadon. WebSocketet exponál egy HTTP-streaming/SSE fallbackkal (mindkettő a centrifugal/centrifuge-on keresztül, ami a Centrifugo és a Grafana Live mögötti library) a /realtime/* alatt, minden kapcsolat JWT-jét a connect frame-ben authentikálja (nincs külön /broadcasting/auth endpoint; v1-ben az egyetlen privát csatorna egy kapcsolat saját subjectje, így elég egyszer, connectkor authentikálni), és minden kapcsolatot server-side feliratkoztat a saját user:<subject> csatornájára. A gyakori esetben nincs szükség kliens-oldali subscribe-logikára.

A megosztott topic-csatornák opt-in és explicit módon authorizáltak: minden kapcsolat JWT-je egy rbac.Identity-ra oldódik fel, és egy topic csak akkor iratkoztatható fel, ha valamelyik Allow/AllowPrefix callback jótáll érte.

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
})

Egy topic, aminek nincs regisztrált Allow callbackje, sosem iratkoztatható fel.

Ha egy csatorna-család teljes neve csak futásidőben ismert (egy dinamikus erőforrás soronként: egy tenant, egy projekt, egy repository), az Allow-val egyenként, induláskor való elnevezés nem lehetséges: helyette AllowPrefix használandó, aminek a callbackje megkapja a teljes topic stringet, hogy meg tudja állapítani, épp melyik konkrét instance-t kérik.

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

Egy pontos Allow-egyezés mindig elsőbbséget élvez egy illeszkedő AllowPrefix-szel szemben; a prefixek közül az elsőként regisztrált, illeszkedő nyer.

Kliens-szerver RPC

A gateway alapból kizárólag szerver→kliens: a kapcsolat a publish-okat kapja, de nem tud írni semmit. A WithRPC bekapcsolja a fordított irányt is: a kliens egy method nevet és egy nyers payloadot küld, a szerver egy regisztrált RPCHandler-rel válaszol, ugyanazzal a JWT-ből feloldott identitással, amit a TopicAuth callbackek is látnak.

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
}))

Egy methodnak, amire nincs regisztrált handler, a kliens centrifuge.ErrorMethodNotFound-ot kap. A handler visszatérési hibája httperr-hez hasonlóan térképeződik le: egy errs.Status-szal ellátott hiba a státusznak megfelelő centrifuge.Error-t adja (400/422 → ErrorBadRequest, 401/403 → ErrorPermissionDenied, 404 → egy framework-definiálta "not found" hiba), errs.Public-ja pedig a kliens felé látható üzenet; egy sima, errs.Status nélküli hiba ErrorInternal-ra esik, belső részlet nélkül.

A gatewaynek nincs adatbázis-hozzáférése (lásd fentebb: "nem hordoz üzleti függőséget"), így egy RPC handler tipikusan egy meglévő HTTP endpointot hív meg (a validáció, a policy és a tranzakció ott van), nem duplikálja az üzleti logikát.

Opciók és a middleware-lánc

A realtime.Run a server.Run-nal azonos stílusú opciókat fogad:

A transport-endpointok körüli framework-lánc a szerver-chassis-éval van összehangolva, ugyanazokból az exportált gpsystem/httpmw építőkockákból, legkülsőtől befelé:

1. CORS, streaming preset (httpmw.CORSStreaming): az API-presettel szemben a credentialök engedélyezettek (egyes streaming-setupok cookie-val autentikálnak), ezért az originenkénti döntést a lentebb leírt origin-ellenőrzés hozza meg wildcard helyett.
2. request id (httpmw.RequestID): minden válasz X-Request-Id fejlécet hordoz, az API-oldallal azonos szemantikával.
3. otel (httpmw.Otel("realtime")): a gateway spanje; a /realtime/healthz ki van szűrve a trace-elésből, WithoutTelemetry mellett pedig a teljes bejegyzés kimarad.
4. Sentry request hub (httpmw.RequestHub): kérésenkénti breadcrumb-izoláció.
5. panic recovery (httpmw.Recoverer): a HTTP-rétegben történő panicból sima 500 lesz, plusz egy stacket hordozó error-log; a gatewaynek nincs Problem-írója, és a streaming kliensek bármely 5xx-re amúgy is újracsatlakoznak.

A WithMiddleware extrák ezeken belül futnak, közvetlenül a transport-endpointok előtt. Access log szándékosan nincs: a kapcsolat-endpointok hosszú életű streameket tartanak, egy kérésenkénti rekord félrevezető lenne.

Skálázás: a példányok szabadon jönnek-mennek

Minden gateway-példány stateless: csak azokat a csatornákat tartja, amikre a saját kapcsolatainak szükségük van, és a közös Valkey broker minden érintett példányhoz eljuttat egy publish-t, bármely processből, sticky session-igény nélkül.

• Felskálázás: docker compose up -d --scale realtime=N; egy új példány nulla állapottal indul, és csak akkor gyűlnek a szubszkripciói, amikor kliensek landolnak rajta.
• Leskálázás / rolling restart: a SIGTERM bekapcsol egy draining flaget (a /realtime/healthz és az új connect-kísérletek 503-at adnak, így a load balancer health checkje kiejti a példányt), a node.Shutdown minden tartott kapcsolatot reconnect-jogosult kóddal zár, a closerek lefutnak, a telemetria flush-öl. A kliens saját reconnect-logikája (backoff, beépítve a JS kliensbe) egy túlélő példányra viszi, ami frissen feliratkoztatja; ami a rés alatt publikálódott, azt a database csatornából állítja helyre, nem a gateway replay-eli.
• Crash: a kliens oldala ugyanaz, mint egy graceful restartnál, csak a drain lépés nélkül.
• Valkey restart: a reconnect és az újra-subscribe a centrifuge dolga; a publikálás a kiesés alatt logolva és eldobva (ugyanaz a fire-and-forget kontraktus), ugyanúgy helyreállítva, mint egy kliens saját reconnectje alatt lemaradt publish.

Config

Lásd az add realtime referenciát a scaffoldolt elemekért (a binary, a compose.yml Traefik-routolt service-e, a .env.example), és az add notification oldalt a broadcast csatorna egy notification Via-jába drótozásához.

Origin-ellenőrzés

A CORS_ORIGINS egyetlen szabályt ad mindhárom transportnak: a same-origin kapcsolatok (az Origin hostja megegyezik a kérés Host fejlécével, a dokumentált production-felállás) és az Origin fejléc nélküli kliensek (szerver-szerver hívások, health check-ek) mindig átmennek, függetlenül a listától; minden más esetben a lista dönt (* mindent enged, egyébként pontos egyezés vagy *-ot tartalmazó minta, pl. https://*.example.com). Idegen origin elutasításakor a gateway Warn szinten logol.

Fejlesztésben gyakori, hogy a frontend és a gateway más originről szolgál ki (pl. egy next dev a hoston, http://localhost:3000, a gateway pedig http://<app>.localhost mögött Traefikkel): ilyenkor a CORS_ORIGINS-nek tartalmaznia kell a frontend originjét (vagy dev módban maradhat *), különben a websocket handshake 403-at kap.

Kliens

import { Centrifuge } from 'centrifuge'

const client = new Centrifuge('wss://your-app.example/realtime/connection/websocket', {
    getToken: async () => fetchAccessToken(), // ugyanaz a JWT, amit az API használ
})
client.on('publication', (ctx) => {
    // ctx.data: { id, name, payload, at }, az id visszaköti a database csatorna sorához
})
client.connect()

A centrifuge-js kezeli a reconnect-backoffot és a token-refresht; a server-side subscription a recipiens saját csatornájára azt jelenti, hogy a kliensnek nincs szüksége explicit subscribe hívásra a neki szóló notificationökhöz. Reconnectkor kérd le az olvasatlan notificationöket a saját modulod inbox-endpointjából (a database csatornára építve), hogy backfilleld, ami a kapcsolat megszakadása alatt lemaradt. A socket egy élő poke, a database az igazságforrás.

Ha a gateway WithRPC-vel regisztrált egy methodot, a kliens client.rpc(...)-vel hívja:

const reply = await client.rpc('chat.send', { body: 'szia' })
// reply.data: a handler nyers válasza, JSON.parse a hívó dolga

Egy sikertelen hívás a centrifuge.Error-t (code, message) dobja: lásd fentebb a "Kliens-szerver RPC" szekciót a hibatérképért.

Kapcsolódó oldalak

• notify/broadcast: a publish-only oldal, amit bármely process importálhat.
• auth/rbac: az Identity, amire egy kapcsolat JWT-je feloldódik, és amit az Allow/AllowPrefix ellenőriz.
• Alkalmazás-életciklus: a közös app.Run váz, amire a realtime.Run épül.