feat(11-03): add the lighthouse realtime package and its Centrifugo driver

- lighthouse: Service/From with realtime.driver selection, RegisterDriver registry, null/log/memory drivers, Route/Surface/Mount, users and actors - centrifugo: HTTP API client (apikey header, 2xx success, no request without a key), five-generator HS256 TokenIssuer, TokenHandler with the WinterCMS 401/503 bodies - module README and root modules table row
2026-09-30 12:18:11 +02:00
parent f1077382f5
commit cada7a4442
11 changed files with 1302 additions and 0 deletions
--- a/modules/lighthouse/centrifugo/client.go
+++ b/modules/lighthouse/centrifugo/client.go
@@ -0,0 +1,203 @@
+package centrifugo
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"strconv"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/wire"
+)
+
+// requestTimeout bounds every Centrifugo API call, as the WinterCMS client
+// does.
+const requestTimeout = 5 * time.Second
+
+// maxResponseBytes caps how much of a response body is read.
+const maxResponseBytes = 1 << 20
+
+// ErrNotConfigured is returned when the secret or key an operation needs
+// is empty: publishing without an API key, or signing without a token
+// secret.
+var ErrNotConfigured = errors.New("centrifugo: not configured")
+
+// Client calls the Centrifugo HTTP API. It is safe for concurrent use.
+type Client struct {
+	apiURL string
+	apiKey string
+	hc     *http.Client
+	log    *slog.Logger
+	now    func() time.Time
+}
+
+// DebugInfo is the connection summary Client.DebugInfo returns. It never
+// carries the API key.
+type DebugInfo struct {
+	APIURL    string `json:"api_url"`
+	Enabled   bool   `json:"enabled"`
+	APIKeySet bool   `json:"api_key_set"`
+}
+
+// NewClient returns a client for cfg.APIURL and cfg.APIKey. hc may be nil
+// for a client with a 5s timeout; every request is also bounded by 5s.
+func NewClient(cfg Config, hc *http.Client) *Client {
+	if hc == nil {
+		hc = &http.Client{Timeout: requestTimeout}
+	}
+	return &Client{
+		apiURL: cfg.APIURL,
+		apiKey: cfg.APIKey,
+		hc:     hc,
+		log:    slog.Default(),
+		now:    time.Now,
+	}
+}
+
+// Enabled reports whether an API key is configured.
+func (c *Client) Enabled() bool { return c != nil && c.apiKey != "" }
+
+// DebugInfo returns the API URL and whether the client is enabled.
+func (c *Client) DebugInfo() DebugInfo {
+	if c == nil {
+		return DebugInfo{}
+	}
+	return DebugInfo{APIURL: c.apiURL, Enabled: c.Enabled(), APIKeySet: c.apiKey != ""}
+}
+
+type eventData struct {
+	Event     string          `json:"event"`
+	Payload   json.RawMessage `json:"payload"`
+	Timestamp wire.Time       `json:"timestamp"`
+}
+
+type publishRequest struct {
+	Channel string    `json:"channel"`
+	Data    eventData `json:"data"`
+}
+
+type broadcastRequest struct {
+	Channels []string  `json:"channels"`
+	Data     eventData `json:"data"`
+}
+
+type presenceRequest struct {
+	Channel string `json:"channel"`
+}
+
+type unsubscribeRequest struct {
+	User    string `json:"user"`
+	Channel string `json:"channel"`
+}
+
+// Publish POSTs {"channel":…,"data":{"event":…,"payload":…,"timestamp":…}}
+// to {api_url}/publish. An empty payload is sent as []. Any 2xx status is
+// success, including Centrifugo's 200 responses that carry an error body.
+func (c *Client) Publish(ctx context.Context, channel, event string, payload json.RawMessage) error {
+	if !c.Enabled() {
+		return ErrNotConfigured
+	}
+	body := publishRequest{Channel: channel, Data: c.data(event, payload)}
+	_, err := c.post(ctx, "/publish", body)
+	return err
+}
+
+// Broadcast POSTs the same data with "channels" to {api_url}/broadcast. No
+// request is sent for an empty channel list.
+func (c *Client) Broadcast(ctx context.Context, channels []string, event string, payload json.RawMessage) error {
+	if !c.Enabled() {
+		return ErrNotConfigured
+	}
+	if len(channels) == 0 {
+		return nil
+	}
+	body := broadcastRequest{Channels: channels, Data: c.data(event, payload)}
+	_, err := c.post(ctx, "/broadcast", body)
+	return err
+}
+
+// Presence returns result.presence of {api_url}/presence for channel. It
+// returns an empty map, never nil, when the client is disabled or the call
+// fails; the error says why.
+func (c *Client) Presence(ctx context.Context, channel string) (map[string]any, error) {
+	out := map[string]any{}
+	if !c.Enabled() {
+		return out, ErrNotConfigured
+	}
+	raw, err := c.post(ctx, "/presence", presenceRequest{Channel: channel})
+	if err != nil {
+		return out, err
+	}
+	var resp struct {
+		Result struct {
+			Presence map[string]any `json:"presence"`
+		} `json:"result"`
+	}
+	if err := json.Unmarshal(raw, &resp); err != nil {
+		c.log.Warn("centrifugo: presence response is not JSON", slog.String("channel", channel))
+		return out, fmt.Errorf("centrifugo: presence: %w", err)
+	}
+	if resp.Result.Presence != nil {
+		out = resp.Result.Presence
+	}
+	return out, nil
+}
+
+// Unsubscribe POSTs {"user":"<id>","channel":…} to {api_url}/unsubscribe.
+func (c *Client) Unsubscribe(ctx context.Context, userID uint, channel string) error {
+	if !c.Enabled() {
+		return ErrNotConfigured
+	}
+	body := unsubscribeRequest{User: strconv.FormatUint(uint64(userID), 10), Channel: channel}
+	_, err := c.post(ctx, "/unsubscribe", body)
+	return err
+}
+
+func (c *Client) data(event string, payload json.RawMessage) eventData {
+	if len(bytes.TrimSpace(payload)) == 0 {
+		payload = json.RawMessage("[]")
+	}
+	return eventData{Event: event, Payload: payload, Timestamp: wire.Time{Time: c.now()}}
+}
+
+// post sends body and returns the response body of a 2xx answer. The API
+// key appears only in the Authorization header, never in logs or errors.
+func (c *Client) post(ctx context.Context, method string, body any) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(body); err != nil {
+		return nil, fmt.Errorf("centrifugo: %s: encode: %w", method, err)
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	ctx, cancel := context.WithTimeout(ctx, requestTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.apiURL+method, bytes.NewReader(bytes.TrimSuffix(buf.Bytes(), []byte("\n"))))
+	if err != nil {
+		return nil, fmt.Errorf("centrifugo: %s: %w", method, err)
+	}
+	req.Header.Set("Authorization", "apikey "+c.apiKey)
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("centrifugo: %s: %w", method, err)
+	}
+	defer resp.Body.Close()
+	raw, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
+		return nil, fmt.Errorf("centrifugo: %s: HTTP %d", method, resp.StatusCode)
+	}
+	if bytes.Contains(raw, []byte(`"error"`)) {
+		// Centrifugo reports API errors as 200 with an error body. The
+		// WinterCMS client counts those as success; so does this one.
+		c.log.Debug("centrifugo: API answered with an error body", slog.String("method", method), slog.Int("status", resp.StatusCode))
+	}
+	return raw, nil
+}
--- a/modules/lighthouse/centrifugo/config.go
+++ b/modules/lighthouse/centrifugo/config.go
@@ -0,0 +1,78 @@
+// Package centrifugo is the Centrifugo driver of lighthouse: an HTTP API
+// client, a connection and subscription token issuer, and the subscribe
+// proxy handler. Import it for its side effect to register the
+// "centrifugo" realtime.driver.
+package centrifugo
+
+import (
+	"strings"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/compass"
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+)
+
+// Default values of the realtime.centrifugo.* keys.
+const (
+	DefaultAPIURL        = "http://127.0.0.1:8001/api"
+	DefaultTokenTTL      = 3600 * time.Second
+	DefaultWSURL         = "/ws"
+	DefaultTokenPath     = "/api/realtime/token"
+	DefaultSubscribePath = "/api/realtime/subscribe"
+)
+
+// Config is the realtime.centrifugo.* configuration.
+type Config struct {
+	// APIURL is the Centrifugo HTTP API base, without a trailing method.
+	APIURL string
+	// APIKey authenticates publishes; empty disables them.
+	APIKey string
+	// TokenSecret signs connection and subscription tokens (HS256); empty
+	// makes the token route answer 503.
+	TokenSecret string
+	// TokenTTL is the lifetime of issued tokens.
+	TokenTTL time.Duration
+	// WSURL is the WebSocket URL the client connects to.
+	WSURL string
+	// ProxySecret is the X-Centrifugo-Secret header value the subscribe
+	// proxy expects; empty denies every subscribe.
+	ProxySecret string
+	// TokenPath and SubscribePath are the mounted route paths.
+	TokenPath     string
+	SubscribePath string
+}
+
+// LoadConfig reads realtime.centrifugo.* from c, filling the defaults.
+// token_ttl is an integer number of seconds or a duration string.
+func LoadConfig(c *compass.Config) Config {
+	cfg := Config{
+		APIURL:        DefaultAPIURL,
+		TokenTTL:      DefaultTokenTTL,
+		WSURL:         DefaultWSURL,
+		TokenPath:     DefaultTokenPath,
+		SubscribePath: DefaultSubscribePath,
+	}
+	if c == nil {
+		return cfg
+	}
+	str := func(key string) string { return strings.TrimSpace(c.String("realtime.centrifugo." + key)) }
+	if v := str("api_url"); v != "" {
+		cfg.APIURL = strings.TrimSuffix(v, "/")
+	}
+	cfg.APIKey = str("api_key")
+	cfg.TokenSecret = str("token_secret")
+	cfg.ProxySecret = str("proxy_secret")
+	if d := lighthouse.DurationSetting(c, "realtime.centrifugo.token_ttl"); d > 0 {
+		cfg.TokenTTL = d
+	}
+	if v := str("ws_url"); v != "" {
+		cfg.WSURL = v
+	}
+	if v := str("token_path"); v != "" {
+		cfg.TokenPath = v
+	}
+	if v := str("subscribe_path"); v != "" {
+		cfg.SubscribePath = v
+	}
+	return cfg
+}
--- a/modules/lighthouse/centrifugo/driver.go
+++ b/modules/lighthouse/centrifugo/driver.go
@@ -0,0 +1,77 @@
+package centrifugo
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+)
+
+// DriverName is the realtime.driver value of this driver.
+const DriverName = "centrifugo"
+
+func init() {
+	lighthouse.RegisterDriver(DriverName, func(app *backpack.App, svc *lighthouse.Service) (lighthouse.Driver, error) {
+		var cfg Config
+		if app != nil {
+			cfg = LoadConfig(app.Config)
+		} else {
+			cfg = LoadConfig(nil)
+		}
+		return NewDriver(svc, cfg, nil), nil
+	})
+}
+
+// Driver is the Centrifugo lighthouse.Driver: it publishes through Client
+// and declares the token route (UserAuth) and the subscribe proxy route
+// (ServerToServer).
+type Driver struct {
+	svc    *lighthouse.Service
+	cfg    Config
+	client *Client
+	issuer *TokenIssuer
+}
+
+// NewDriver builds the driver for svc from cfg. hc may be nil (see
+// NewClient).
+func NewDriver(svc *lighthouse.Service, cfg Config, hc *http.Client) *Driver {
+	c := NewClient(cfg, hc)
+	c.log = svc.Logger()
+	return &Driver{
+		svc:    svc,
+		cfg:    cfg,
+		client: c,
+		issuer: NewTokenIssuer(cfg.TokenSecret, cfg.TokenTTL),
+	}
+}
+
+// Name returns "centrifugo".
+func (d *Driver) Name() string { return DriverName }
+
+// Config returns the driver's configuration.
+func (d *Driver) Config() Config { return d.cfg }
+
+// Client returns the HTTP API client.
+func (d *Driver) Client() *Client { return d.client }
+
+// Issuer returns the token issuer.
+func (d *Driver) Issuer() *TokenIssuer { return d.issuer }
+
+// Publish sends event to one channel through Client.Publish.
+func (d *Driver) Publish(ctx context.Context, channel, event string, payload json.RawMessage) error {
+	return d.client.Publish(ctx, channel, event, payload)
+}
+
+// Broadcast sends event to several channels through Client.Broadcast.
+func (d *Driver) Broadcast(ctx context.Context, channels []string, event string, payload json.RawMessage) error {
+	return d.client.Broadcast(ctx, channels, event, payload)
+}
+
+// Routes returns GET token_path (UserAuth, TokenHandler).
+func (d *Driver) Routes() []lighthouse.Route {
+	return []lighthouse.Route{
+		{Name: "token", Method: http.MethodGet, Path: d.cfg.TokenPath, Surface: lighthouse.UserAuth, Handler: TokenHandler(d.svc, d.issuer)},
+	}
+}
--- a/modules/lighthouse/centrifugo/handlers.go
+++ b/modules/lighthouse/centrifugo/handlers.go
@@ -0,0 +1,67 @@
+package centrifugo
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+
+	"git.golem15.com/golem15/summercms/modules/bouncer"
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+)
+
+type errorBody struct {
+	Error string `json:"error"`
+}
+
+type tokenBody struct {
+	Token string `json:"token"`
+}
+
+// TokenHandler issues the connection token of the signed-in user. It must
+// be mounted behind a user guard (the UserAuth surface).
+//
+//   - no principal, or no user for it: 401 {"error":"Unauthorized"}
+//   - no token secret: 503 with the WinterCMS not-configured error body
+//   - otherwise 200 {"token":"…"} (see TokenIssuer.ForUser)
+func TokenHandler(svc *lighthouse.Service, issuer *TokenIssuer) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		p, ok := bouncer.User(r.Context())
+		if !ok || p == nil || p.ID == 0 {
+			writeJSON(w, http.StatusUnauthorized, errorBody{Error: "Unauthorized"})
+			return
+		}
+		u, found, err := svc.User(r.Context(), p.ID)
+		if err != nil || !found {
+			writeJSON(w, http.StatusUnauthorized, errorBody{Error: "Unauthorized"})
+			return
+		}
+		if !issuer.Configured() {
+			writeJSON(w, http.StatusServiceUnavailable, errorBody{Error: "WebSocket not configured"})
+			return
+		}
+		tok, err := issuer.ForUser(u)
+		if err != nil {
+			svc.Logger().Error("realtime: token signing failed", "error", err)
+			writeJSON(w, http.StatusInternalServerError, errorBody{Error: "Internal server error"})
+			return
+		}
+		writeJSON(w, http.StatusOK, tokenBody{Token: tok})
+	}
+}
+
+// writeJSON writes v with no trailing newline and no HTML escaping, plus the
+// Content-Type and Cache-Control headers of a Laravel JSON response.
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	h := w.Header()
+	h.Set("Content-Type", "application/json")
+	h.Set("Cache-Control", "no-cache, private")
+	w.WriteHeader(status)
+	_, _ = w.Write(bytes.TrimSuffix(buf.Bytes(), []byte("\n")))
+}
--- a/modules/lighthouse/centrifugo/token.go
+++ b/modules/lighthouse/centrifugo/token.go
@@ -0,0 +1,145 @@
+package centrifugo
+
+import (
+	"bytes"
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/lighthouse"
+	"github.com/golang-jwt/jwt/v5"
+)
+
+// anonymousTTL is the lifetime of Anonymous tokens.
+const anonymousTTL = 300 * time.Second
+
+// TokenIssuer signs Centrifugo connection and subscription tokens with
+// HS256. Its claims match the WinterCMS JwtTokenGenerator. It is safe for
+// concurrent use.
+type TokenIssuer struct {
+	secret []byte
+	ttl    time.Duration
+	// Now is the clock used for exp; nil means time.Now.
+	Now func() time.Time
+}
+
+// NewTokenIssuer returns an issuer for secret with tokens valid for ttl
+// (DefaultTokenTTL when ttl is not positive).
+func NewTokenIssuer(secret string, ttl time.Duration) *TokenIssuer {
+	if ttl <= 0 {
+		ttl = DefaultTokenTTL
+	}
+	return &TokenIssuer{secret: []byte(secret), ttl: ttl}
+}
+
+// Configured reports whether a signing secret is set.
+func (i *TokenIssuer) Configured() bool { return i != nil && len(i.secret) > 0 }
+
+type userInfo struct {
+	Name *string `json:"name"`
+}
+
+type userClaims struct {
+	Sub  string   `json:"sub"`
+	Exp  int64    `json:"exp"`
+	Info userInfo `json:"info"`
+}
+
+type channelClaims struct {
+	Sub     string `json:"sub"`
+	Channel string `json:"channel"`
+	Exp     int64  `json:"exp"`
+}
+
+type anonymousClaims struct {
+	Sub string `json:"sub"`
+	Exp int64  `json:"exp"`
+}
+
+type identifierClaims struct {
+	Sub  string          `json:"sub"`
+	Exp  int64           `json:"exp"`
+	Info json.RawMessage `json:"info"`
+}
+
+// ForUser returns a connection token with exactly the claims sub (the user
+// id as a string), exp (now + TTL) and info {"name": u.Name}. info carries
+// nothing else: no email, no other ids.
+func (i *TokenIssuer) ForUser(u lighthouse.User) (string, error) {
+	return i.sign(userClaims{Sub: userSub(u.ID), Exp: i.exp(i.ttl), Info: userInfo{Name: u.Name}})
+}
+
+// Subscription returns a subscription token with the claims sub, channel
+// and exp.
+func (i *TokenIssuer) Subscription(u lighthouse.User, channel string) (string, error) {
+	return i.sign(channelClaims{Sub: userSub(u.ID), Channel: channel, Exp: i.exp(i.ttl)})
+}
+
+// Anonymous returns a connection token with sub "" and exp now + 5 minutes.
+func (i *TokenIssuer) Anonymous() (string, error) {
+	return i.sign(anonymousClaims{Sub: "", Exp: i.exp(anonymousTTL)})
+}
+
+// ForIdentifier returns a connection token for a non-user identifier with
+// the claims sub, exp and info. An empty info is encoded as [], as the
+// WinterCMS generator's empty PHP array is.
+func (i *TokenIssuer) ForIdentifier(identifier string, info map[string]any) (string, error) {
+	raw := json.RawMessage("[]")
+	if len(info) > 0 {
+		b, err := marshal(info)
+		if err != nil {
+			return "", err
+		}
+		raw = b
+	}
+	return i.sign(identifierClaims{Sub: identifier, Exp: i.exp(i.ttl), Info: raw})
+}
+
+// SubscriptionForIdentifier returns a subscription token for a non-user
+// identifier with the claims sub, channel and exp.
+func (i *TokenIssuer) SubscriptionForIdentifier(identifier, channel string) (string, error) {
+	return i.sign(channelClaims{Sub: identifier, Channel: channel, Exp: i.exp(i.ttl)})
+}
+
+func (i *TokenIssuer) exp(ttl time.Duration) int64 {
+	now := time.Now
+	if i != nil && i.Now != nil {
+		now = i.Now
+	}
+	return now().Add(ttl).Unix()
+}
+
+func (i *TokenIssuer) sign(claims any) (string, error) {
+	if !i.Configured() {
+		return "", ErrNotConfigured
+	}
+	raw, err := marshal(claims)
+	if err != nil {
+		return "", err
+	}
+	return jwt.NewWithClaims(jwt.SigningMethodHS256, orderedClaims(raw)).SignedString(i.secret)
+}
+
+func userSub(id uint) string { return strconv.FormatUint(uint64(id), 10) }
+
+func marshal(v any) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		return nil, err
+	}
+	return bytes.TrimSuffix(buf.Bytes(), []byte("\n")), nil
+}
+
+// orderedClaims keeps the claim order of the struct it was marshalled
+// from. The jwt.Claims methods are never used for signing.
+type orderedClaims json.RawMessage
+
+func (c orderedClaims) MarshalJSON() ([]byte, error)               { return c, nil }
+func (orderedClaims) GetExpirationTime() (*jwt.NumericDate, error) { return nil, nil }
+func (orderedClaims) GetIssuedAt() (*jwt.NumericDate, error)       { return nil, nil }
+func (orderedClaims) GetNotBefore() (*jwt.NumericDate, error)      { return nil, nil }
+func (orderedClaims) GetIssuer() (string, error)                   { return "", nil }
+func (orderedClaims) GetSubject() (string, error)                  { return "", nil }
+func (orderedClaims) GetAudience() (jwt.ClaimStrings, error)       { return nil, nil }