feat: add private r2 bucket

.env.example

@@ -24,6 +24,11 @@ R2_SECRET_ACCESS_KEY=
 R2_BUCKET=
 # Public base URL where bucket objects are served (custom domain or pub-XXX.r2.dev)
 R2_PUBLIC_URL=
+# Separate bucket for private media (DM attachments), read only via short-lived
+# presigned GETs. It must have NO public access and no custom domain: R2 public
+# access is per-bucket, so a prefix inside R2_BUCKET would still be world-readable.
+# Empty disables private-media endpoints (they return 503).
+R2_PRIVATE_BUCKET=
 
 # SMTP (password-reset + email-verification + /feedback admin notifications)
 SMTP_HOST=

internal/config/config.go

@@ -37,6 +37,13 @@ type Config struct {
 	R2Bucket          string
 	R2PublicURL       string // Public base URL for objects, e.g. https://cdn.example.com or the R2 public bucket URL
 
+	// R2PrivateBucket is a SEPARATE bucket with no public access, used for
+	// objects that must never be fetchable by URL alone (DM attachments).
+	// A prefix inside R2Bucket would not do: R2 public access is per-bucket,
+	// so anything in that bucket is world-readable. Empty disables the
+	// private-media endpoints (they return 503) — the BFF still boots.
+	R2PrivateBucket string
+
 	// SMTP — used for password-reset, email-verification, and
 	// /feedback admin-notification mails.
 	SMTPHost     string
@@ -136,6 +143,7 @@ func Load() (*Config, error) {
 		R2SecretAccessKey: getEnv("R2_SECRET_ACCESS_KEY", ""),
 		R2Bucket:          getEnv("R2_BUCKET", ""),
 		R2PublicURL:       getEnv("R2_PUBLIC_URL", ""),
+		R2PrivateBucket:   getEnv("R2_PRIVATE_BUCKET", ""),
 
 		SMTPHost:     getEnv("SMTP_HOST", ""),
 		SMTPPort:     getEnvInt("SMTP_PORT", 587),

internal/handler/broadcasting_auth.go

@@ -0,0 +1,174 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strings"
+
+	"github.com/prabhupadasays/spsays-bff/internal/middleware"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
+	"github.com/prabhupadasays/spsays-bff/pkg/response"
+	"go.uber.org/zap"
+)
+
+// ChannelRule authorizes one family of channels.
+//
+// Suffix is whatever follows Prefix in the channel name (a user ID today, a
+// conversation ID once chat lands). Authorize reports whether userID — the
+// authenticated caller — may subscribe. Returning an error means "couldn't
+// decide" (a lookup failed) and surfaces as a 500; returning (false, nil)
+// means "decided: no" and surfaces as a 403.
+type ChannelRule struct {
+	Prefix    string
+	Authorize func(ctx context.Context, userID, suffix string) (bool, error)
+}
+
+// BroadcastingAuthHandler implements the Pusher channel-authorization
+// endpoint that pusher-js calls (as `authEndpoint`) before subscribing to a
+// private- or presence- channel.
+//
+// This is the gate that makes private channels private: Soketi will not honour
+// a subscription without a signature, and the signature is only minted here,
+// after RequireAuth has established who the caller is and a rule has agreed
+// they may listen. Unknown channels are denied by default, so adding a channel
+// family is an explicit act.
+type BroadcastingAuthHandler struct {
+	rt     *realtime.SoketiClient
+	rules  []ChannelRule
+	logger *zap.Logger
+}
+
+func NewBroadcastingAuthHandler(rt *realtime.SoketiClient, logger *zap.Logger, rules ...ChannelRule) *BroadcastingAuthHandler {
+	return &BroadcastingAuthHandler{rt: rt, rules: rules, logger: logger}
+}
+
+// NotificationsRule allows a user to subscribe to their own notification
+// channel and nobody else's.
+func NotificationsRule() ChannelRule {
+	return ChannelRule{
+		Prefix: "private-notifications.",
+		Authorize: func(_ context.Context, userID, suffix string) (bool, error) {
+			return suffix != "" && suffix == userID, nil
+		},
+	}
+}
+
+type broadcastingAuthResp struct {
+	Auth        string `json:"auth"`
+	ChannelData string `json:"channel_data,omitempty"`
+}
+
+// writeAuth emits the auth payload BARE, deliberately bypassing the API's
+// {success, data} envelope.
+//
+// This endpoint speaks Pusher's protocol, not our API's: pusher-js reads
+// `auth` off the top level of the parsed body, so an enveloped response makes
+// every private subscription fail. Errors still use response.Error — the
+// client only checks the status code on that path.
+func writeAuth(w http.ResponseWriter, payload broadcastingAuthResp) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(w).Encode(payload)
+}
+
+// Handle — POST /api/v1/broadcasting/auth. Session-authenticated (the cookie
+// travels because the frontend sets credentials: "include").
+//
+// pusher-js posts application/x-www-form-urlencoded by default; JSON is
+// accepted too so a custom client isn't forced into form encoding.
+func (h *BroadcastingAuthHandler) Handle(w http.ResponseWriter, r *http.Request) {
+	userID := middleware.UserID(r.Context())
+	if userID == "" {
+		// RequireAuth should have caught this; refuse rather than sign.
+		response.Error(w, http.StatusUnauthorized, "authentication required")
+		return
+	}
+
+	socketID, channel := parseBroadcastingAuthReq(r)
+	if socketID == "" || channel == "" {
+		response.Error(w, http.StatusBadRequest, "socket_id and channel_name are required")
+		return
+	}
+
+	rule, suffix, found := h.match(channel)
+	if !found {
+		// Default-deny. An unrecognised channel is a client bug or an
+		// probe — either way there's nothing to sign.
+		h.logger.Warn("broadcasting auth: unknown channel",
+			zap.String("channel", channel), zap.String("user_id", userID))
+		response.Error(w, http.StatusForbidden, "channel not available")
+		return
+	}
+
+	allowed, err := rule.Authorize(r.Context(), userID, suffix)
+	if err != nil {
+		h.logger.Error("broadcasting auth: authorize failed",
+			zap.String("channel", channel), zap.Error(err))
+		response.Error(w, http.StatusInternalServerError, "could not authorize channel")
+		return
+	}
+	if !allowed {
+		h.logger.Warn("broadcasting auth: denied",
+			zap.String("channel", channel), zap.String("user_id", userID))
+		response.Error(w, http.StatusForbidden, "not permitted on this channel")
+		return
+	}
+
+	if strings.HasPrefix(channel, "presence-") {
+		auth, channelData, err := h.rt.AuthorizePresence(socketID, channel, realtime.PresenceData{UserID: userID})
+		if err != nil {
+			h.authError(w, err)
+			return
+		}
+		writeAuth(w, broadcastingAuthResp{Auth: auth, ChannelData: channelData})
+		return
+	}
+
+	auth, err := h.rt.AuthorizeChannel(socketID, channel)
+	if err != nil {
+		h.authError(w, err)
+		return
+	}
+	writeAuth(w, broadcastingAuthResp{Auth: auth})
+}
+
+// match finds the rule owning this channel and returns the part of the name
+// after its prefix.
+func (h *BroadcastingAuthHandler) match(channel string) (ChannelRule, string, bool) {
+	for _, rule := range h.rules {
+		if strings.HasPrefix(channel, rule.Prefix) {
+			return rule, strings.TrimPrefix(channel, rule.Prefix), true
+		}
+	}
+	return ChannelRule{}, "", false
+}
+
+func (h *BroadcastingAuthHandler) authError(w http.ResponseWriter, err error) {
+	if errors.Is(err, realtime.ErrNotConfigured) {
+		response.Error(w, http.StatusServiceUnavailable, "realtime is not configured")
+		return
+	}
+	h.logger.Error("broadcasting auth: signing failed", zap.Error(err))
+	response.Error(w, http.StatusBadRequest, err.Error())
+}
+
+// parseBroadcastingAuthReq reads socket_id / channel_name from either a form
+// body (pusher-js default) or a JSON body.
+func parseBroadcastingAuthReq(r *http.Request) (socketID, channel string) {
+	if strings.HasPrefix(r.Header.Get("Content-Type"), "application/json") {
+		var body struct {
+			SocketID    string `json:"socket_id"`
+			ChannelName string `json:"channel_name"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			return "", ""
+		}
+		return strings.TrimSpace(body.SocketID), strings.TrimSpace(body.ChannelName)
+	}
+	if err := r.ParseForm(); err != nil {
+		return "", ""
+	}
+	return strings.TrimSpace(r.PostFormValue("socket_id")), strings.TrimSpace(r.PostFormValue("channel_name"))
+}

internal/handler/broadcasting_auth_test.go

@@ -0,0 +1,152 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/prabhupadasays/spsays-bff/internal/middleware"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
+	"go.uber.org/zap"
+)
+
+const testUser = "user-a"
+
+func testAuthHandler(t *testing.T) *BroadcastingAuthHandler {
+	t.Helper()
+	rt := realtime.NewSoketiClient("app-1", "test-key", "test-secret", "ws.example.com", 443, true, zap.NewNop())
+	if rt == nil {
+		t.Fatal("realtime client is nil")
+	}
+	// A second rule stands in for the chat conversation channels, proving the
+	// rule table is what decides access rather than anything hardcoded.
+	convRule := ChannelRule{
+		Prefix: "presence-conversation.",
+		Authorize: func(_ context.Context, userID, suffix string) (bool, error) {
+			return suffix == "open-to-user-a" && userID == testUser, nil
+		},
+	}
+	return NewBroadcastingAuthHandler(rt, zap.NewNop(), NotificationsRule(), convRule)
+}
+
+// serve posts a form-encoded auth request as userID ("" = unauthenticated).
+func serve(t *testing.T, h *BroadcastingAuthHandler, userID, socketID, channel string) *httptest.ResponseRecorder {
+	t.Helper()
+	form := url.Values{"socket_id": {socketID}, "channel_name": {channel}}
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/broadcasting/auth", strings.NewReader(form.Encode()))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	if userID != "" {
+		req = req.WithContext(middleware.WithUserID(req.Context(), userID))
+	}
+	rec := httptest.NewRecorder()
+	h.Handle(rec, req)
+	return rec
+}
+
+func TestBroadcastingAuth_Authorization(t *testing.T) {
+	h := testAuthHandler(t)
+
+	tests := []struct {
+		name    string
+		userID  string
+		channel string
+		want    int
+	}{
+		{"own notifications channel", testUser, "private-notifications." + testUser, http.StatusOK},
+		{"another user's notifications", testUser, "private-notifications.user-b", http.StatusForbidden},
+		{"unauthenticated", "", "private-notifications." + testUser, http.StatusUnauthorized},
+		{"unknown channel family", testUser, "private-secrets.42", http.StatusForbidden},
+		{"legacy public channel is not signable", testUser, "notifications." + testUser, http.StatusForbidden},
+		{"empty suffix", testUser, "private-notifications.", http.StatusForbidden},
+		{"presence channel allowed by rule", testUser, "presence-conversation.open-to-user-a", http.StatusOK},
+		{"presence channel denied by rule", testUser, "presence-conversation.someone-elses", http.StatusForbidden},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			rec := serve(t, h, tc.userID, "1234.5678", tc.channel)
+			if rec.Code != tc.want {
+				t.Errorf("status = %d, want %d (body %s)", rec.Code, tc.want, rec.Body.String())
+			}
+		})
+	}
+}
+
+// A prefix match must not be mistaken for authorization: "…user-a-evil" starts
+// with the victim's id but is a different channel.
+func TestBroadcastingAuth_SuffixIsComparedExactly(t *testing.T) {
+	rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "private-notifications."+testUser+"-evil")
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("status = %d, want 403", rec.Code)
+	}
+}
+
+func TestBroadcastingAuth_SignatureShape(t *testing.T) {
+	rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "private-notifications."+testUser)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", rec.Code)
+	}
+	var body struct {
+		Auth        string `json:"auth"`
+		ChannelData string `json:"channel_data"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if !strings.HasPrefix(body.Auth, "test-key:") {
+		t.Errorf("auth = %q, want a <key>:<signature> pair", body.Auth)
+	}
+	// channel_data belongs to presence channels only; sending it for a private
+	// channel would make Soketi reject the subscription.
+	if body.ChannelData != "" {
+		t.Errorf("channel_data = %q, want empty for a private channel", body.ChannelData)
+	}
+}
+
+func TestBroadcastingAuth_PresenceReturnsChannelData(t *testing.T) {
+	rec := serve(t, testAuthHandler(t), testUser, "1234.5678", "presence-conversation.open-to-user-a")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", rec.Code)
+	}
+	var body struct {
+		ChannelData string `json:"channel_data"`
+	}
+	_ = json.Unmarshal(rec.Body.Bytes(), &body)
+	if !strings.Contains(body.ChannelData, `"user_id":"`+testUser+`"`) {
+		t.Errorf("channel_data = %q, want the caller's user_id", body.ChannelData)
+	}
+}
+
+func TestBroadcastingAuth_AcceptsJSONBody(t *testing.T) {
+	body := `{"socket_id":"1234.5678","channel_name":"private-notifications.` + testUser + `"}`
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/broadcasting/auth", strings.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	req = req.WithContext(middleware.WithUserID(req.Context(), testUser))
+	rec := httptest.NewRecorder()
+	testAuthHandler(t).Handle(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Errorf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBroadcastingAuth_MissingParams(t *testing.T) {
+	h := testAuthHandler(t)
+	if rec := serve(t, h, testUser, "", "private-notifications."+testUser); rec.Code != http.StatusBadRequest {
+		t.Errorf("missing socket_id: status = %d, want 400", rec.Code)
+	}
+	if rec := serve(t, h, testUser, "1234.5678", ""); rec.Code != http.StatusBadRequest {
+		t.Errorf("missing channel_name: status = %d, want 400", rec.Code)
+	}
+}
+
+// With realtime disabled there is no secret to sign with; the endpoint must
+// say so rather than hand back an unusable signature.
+func TestBroadcastingAuth_UnconfiguredRealtime(t *testing.T) {
+	h := NewBroadcastingAuthHandler(nil, zap.NewNop(), NotificationsRule())
+	if rec := serve(t, h, testUser, "1234.5678", "private-notifications."+testUser); rec.Code != http.StatusServiceUnavailable {
+		t.Errorf("status = %d, want 503", rec.Code)
+	}
+}

internal/handler/soketi_webhook.go

@@ -10,14 +10,26 @@ import (
 	"strings"
 
 	"github.com/prabhupadasays/spsays-bff/internal/repository"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
 	"go.uber.org/zap"
 )
 
-// notifChannelPrefix is the public per-user channel the frontend subscribes to
+// userIDFromNotifChannel extracts the user id from a per-user notifications
-// for realtime notifications. Because it's keyed by user, a `channel_vacated`
+// channel, matching both the private name and the legacy public one.
-// webhook on it means that user's *last* connection dropped — i.e. they went
+//
-// offline (across all their tabs) — so we can evict their feed seen-set.
+// Because these channels are keyed by user, a `channel_vacated` webhook on one
-const notifChannelPrefix = "notifications."
+// means that user's *last* connection dropped — i.e. they went offline across
+// all their tabs — so we can evict their feed seen-set. Both names are matched
+// for the duration of the private-channel migration: clients running a
+// service-worker-cached bundle still sit on the legacy channel, and matching
+// only one name would silently stop evicting for half the users.
+func userIDFromNotifChannel(channel string) (string, bool) {
+	if uid, ok := strings.CutPrefix(channel, realtime.UserChannelPrefix); ok {
+		return uid, uid != ""
+	}
+	uid, ok := strings.CutPrefix(channel, realtime.LegacyUserChannelPrefix)
+	return uid, ok && uid != ""
+}
 
 // SoketiWebhookHandler receives Soketi (Pusher-compatible) webhooks. Soketi
 // signs each delivery with the app secret (X-Pusher-Signature = hex
@@ -83,7 +95,7 @@ func (h *SoketiWebhookHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		if e.Name != "channel_vacated" {
 			continue
 		}
-		if uid, ok := strings.CutPrefix(e.Channel, notifChannelPrefix); ok && uid != "" {
+		if uid, ok := userIDFromNotifChannel(e.Channel); ok {
 			h.feed.Reset(ctx, uid)
 			cleared++
 		}

internal/handler/soketi_webhook_test.go

@@ -0,0 +1,31 @@
+package handler
+
+import "testing"
+
+// Both channel names must resolve during the private-channel migration:
+// matching only one would silently stop evicting feed seen-sets for every
+// client still on the other name.
+func TestUserIDFromNotifChannel(t *testing.T) {
+	tests := []struct {
+		channel string
+		wantID  string
+		wantOK  bool
+	}{
+		{"private-notifications.user-a", "user-a", true},
+		{"notifications.user-a", "user-a", true},
+		{"private-notifications.", "", false},
+		{"notifications.", "", false},
+		{"presence-conversation.c1", "", false},
+		{"", "", false},
+		// Guards against a naive suffix split: the private name must not be
+		// mistaken for the legacy one and yield a mangled id.
+		{"private-notifications.notifications.x", "notifications.x", true},
+	}
+	for _, tc := range tests {
+		gotID, gotOK := userIDFromNotifChannel(tc.channel)
+		if gotOK != tc.wantOK || (tc.wantOK && gotID != tc.wantID) {
+			t.Errorf("userIDFromNotifChannel(%q) = (%q, %v), want (%q, %v)",
+				tc.channel, gotID, gotOK, tc.wantID, tc.wantOK)
+		}
+	}
+}

internal/middleware/auth.go

@@ -100,6 +100,13 @@ func OptionalAuth(issuer *auth.Issuer, users *repository.UserRepo) func(http.Han
 	}
 }
 
+// WithUserID returns a context carrying userID, as RequireAuth would set it.
+// The exported counterpart to UserID: it lets handlers be exercised in tests
+// without standing up a JWT issuer and a user repository.
+func WithUserID(ctx context.Context, userID string) context.Context {
+	return context.WithValue(ctx, userIDKey, userID)
+}
+
 // UserID returns the authenticated user's id from the request context.
 // Returns "" if RequireAuth has not run.
 func UserID(ctx context.Context) string {

internal/repository/r2.go

@@ -25,6 +25,11 @@ type R2 struct {
 	client     *s3.Client
 	bucket     string
 	publicBase string // R2_PUBLIC_URL, no trailing slash
+
+	// privateBucket is a separate, non-public bucket for objects that must
+	// only be reachable through a presigned GET (see PresignGet). Empty when
+	// unconfigured, which makes PrivateConfigured report false.
+	privateBucket string
 }
 
 // R2Config captures the env vars needed to talk to R2.
@@ -34,6 +39,9 @@ type R2Config struct {
 	SecretAccessKey string
 	Bucket          string
 	PublicURL       string
+	// PrivateBucket is optional: absent means private media is disabled,
+	// while the public bucket keeps working.
+	PrivateBucket string
 }
 
 // Configured returns true if every required field is present.
@@ -61,9 +69,10 @@ func NewR2(cfg R2Config) (*R2, error) {
 	})
 
 	return &R2{
-		client:     client,
+		client:        client,
-		bucket:     cfg.Bucket,
+		bucket:        cfg.Bucket,
-		publicBase: trimTrailingSlash(cfg.PublicURL),
+		publicBase:    trimTrailingSlash(cfg.PublicURL),
+		privateBucket: cfg.PrivateBucket,
 	}, nil
 }
 
@@ -129,3 +138,91 @@ func trimTrailingSlash(s string) string {
 	}
 	return s
 }
+
+// ---------------------------------------------------------------------------
+// Private media
+//
+// Objects in the public bucket are world-readable by URL — fine for avatars
+// and post images, wrong for anything private (DM attachments). R2 public
+// access is per-bucket, not per-prefix, so private objects live in a
+// SEPARATE bucket that has no public access and no custom domain. The only
+// way to read them is a short-lived presigned GET, which the BFF issues after
+// checking that the caller is allowed to see that particular object.
+// ---------------------------------------------------------------------------
+
+// PrivateConfigured reports whether the private bucket is available. Callers
+// use it to return 503 rather than silently falling back to public storage —
+// a fallback there would leak the very objects this bucket exists to protect.
+func (r *R2) PrivateConfigured() bool {
+	return r != nil && r.privateBucket != ""
+}
+
+// PresignPutPrivate returns a one-shot PUT URL for the PRIVATE bucket.
+//
+// Unlike PresignPut it deliberately returns no public URL: there isn't one,
+// and handing callers a URL-shaped string for a private object invites it
+// being persisted and treated as fetchable. Store the key instead, and mint a
+// read URL per request with PresignGet.
+func (r *R2) PresignPutPrivate(ctx context.Context, key, contentType string, expiry time.Duration) (uploadURL string, err error) {
+	if !r.PrivateConfigured() {
+		return "", errors.New("private media storage is not configured")
+	}
+	if key == "" {
+		return "", errors.New("empty key")
+	}
+
+	presigner := s3.NewPresignClient(r.client)
+	req, err := presigner.PresignPutObject(ctx, &s3.PutObjectInput{
+		Bucket:      aws.String(r.privateBucket),
+		Key:         aws.String(key),
+		ContentType: aws.String(contentType),
+	}, s3.WithPresignExpires(expiry))
+	if err != nil {
+		return "", fmt.Errorf("presign private put: %w", err)
+	}
+	return req.URL, nil
+}
+
+// PresignGet returns a short-lived read URL for an object in the private
+// bucket.
+//
+// Keep expiry short. The URL is a bearer credential for that one object:
+// anyone holding it can read until it expires, so it should be minted per
+// request and never persisted in a message payload or an offline cache.
+func (r *R2) PresignGet(ctx context.Context, key string, expiry time.Duration) (string, error) {
+	if !r.PrivateConfigured() {
+		return "", errors.New("private media storage is not configured")
+	}
+	if key == "" {
+		return "", errors.New("empty key")
+	}
+
+	presigner := s3.NewPresignClient(r.client)
+	req, err := presigner.PresignGetObject(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(r.privateBucket),
+		Key:    aws.String(key),
+	}, s3.WithPresignExpires(expiry))
+	if err != nil {
+		return "", fmt.Errorf("presign get: %w", err)
+	}
+	return req.URL, nil
+}
+
+// DeletePrivate removes an object from the private bucket. Retention and
+// account deletion need real erasure, not a tombstone row that leaves the
+// bytes sitting in storage.
+func (r *R2) DeletePrivate(ctx context.Context, key string) error {
+	if !r.PrivateConfigured() {
+		return errors.New("private media storage is not configured")
+	}
+	if key == "" {
+		return errors.New("empty key")
+	}
+	if _, err := r.client.DeleteObject(ctx, &s3.DeleteObjectInput{
+		Bucket: aws.String(r.privateBucket),
+		Key:    aws.String(key),
+	}); err != nil {
+		return fmt.Errorf("delete private object %q: %w", key, err)
+	}
+	return nil
+}

internal/server/server.go

@@ -68,6 +68,7 @@ type Server struct {
 	profileH       *handler.ProfileHandler
 	relationH      *handler.RelationshipHandler
 	soketiWebhookH *handler.SoketiWebhookHandler
+	broadcastAuthH *handler.BroadcastingAuthHandler
 	notificationH  *handler.NotificationHandler
 	sadhnaH        *handler.SadhnaHandler
 	pushH          *handler.PushHandler
@@ -157,6 +158,7 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 		SecretAccessKey: cfg.R2SecretAccessKey,
 		Bucket:          cfg.R2Bucket,
 		PublicURL:       cfg.R2PublicURL,
+		PrivateBucket:   cfg.R2PrivateBucket,
 	})
 	if err != nil {
 		logger.Warn("auth: R2 unavailable — avatar uploads will return 503", zap.Error(err))
@@ -256,6 +258,12 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 		cfg.SoketiAppID, cfg.SoketiAppKey, cfg.SoketiAppSecret,
 		cfg.SoketiHost, cfg.SoketiPort, cfg.SoketiUseTLS, logger,
 	)
+	// Channel authorization for private-* subscriptions. Rules are explicit
+	// and default-deny; chat conversation channels get added here.
+	s.broadcastAuthH = handler.NewBroadcastingAuthHandler(rtClient, logger,
+		handler.NotificationsRule(),
+	)
+
 	pushRepo := repository.NewPushRepo(s.pg)
 	pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger)
 	s.pushH = handler.NewPushHandler(pushSender, pushRepo)
@@ -887,6 +895,17 @@ func (s *Server) setupRouter() {
 			r.Post("/webhooks/soketi", s.soketiWebhookH.Handle)
 		}
 
+		// Pusher channel authorization — pusher-js posts here (as its
+		// `authEndpoint`) before subscribing to a private-/presence- channel.
+		// Session-authenticated: the signature is only minted for a caller
+		// who has proven who they are and passed the channel's rule.
+		if s.broadcastAuthH != nil {
+			r.Group(func(r chi.Router) {
+				r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo))
+				r.Post("/broadcasting/auth", s.broadcastAuthH.Handle)
+			})
+		}
+
 		// Home feed — the global, personalised feed of posts + its "new on
 		// the platform" discovery strip. Auth-required (the feed is
 		// per-viewer: ordering + relationship flags depend on the caller).

internal/service/notification_service.go

@@ -276,7 +276,7 @@ func (s *NotificationService) NotifyPostLike(ctx context.Context, postID, author
 			payload["actor_handle"] = actorUser.Handle
 			payload["actor_avatar_url"] = actorUser.AvatarURL
 		}
-		if err := s.rt.Trigger(ctx, "notifications."+authorID, "notification", payload); err != nil {
+		if err := s.rt.TriggerUser(ctx, authorID, "notification", payload); err != nil {
 			s.logger.Warn("realtime like publish failed", zap.String("recipient", authorID), zap.Error(err))
 		}
 	}
@@ -534,7 +534,7 @@ func (s *NotificationService) publishRealtime(ctx context.Context, recipientID,
 		payload["actor_handle"] = actor.Handle
 		payload["actor_avatar_url"] = actor.AvatarURL
 	}
-	if err := s.rt.Trigger(ctx, "notifications."+recipientID, "notification", payload); err != nil {
+	if err := s.rt.TriggerUser(ctx, recipientID, "notification", payload); err != nil {
 		s.logger.Warn("realtime notification publish failed",
 			zap.String("recipient", recipientID), zap.String("kind", kind), zap.Error(err))
 	}
@@ -550,7 +550,7 @@ func (s *NotificationService) SendReminder(ctx context.Context, userID, kind, ti
 	// Realtime toast — a distinct "reminder" event so the client toasts
 	// without bumping the unread badge (reminders have no DB row).
 	if s.rt != nil && !prefOff(off, kind, repository.ChannelInApp) {
-		_ = s.rt.Trigger(ctx, "notifications."+userID, "reminder", map[string]any{
+		_ = s.rt.TriggerUser(ctx, userID, "reminder", map[string]any{
 			"kind": kind, "title": title, "body": body, "url": url,
 		})
 	}
@@ -632,7 +632,7 @@ func (s *NotificationService) NotifyAdminMessage(ctx context.Context, recipientI
 	url := s.frontendURL + "/notifications"
 
 	if s.rt != nil {
-		if err := s.rt.Trigger(ctx, "notifications."+recipientID, "notification", map[string]any{
+		if err := s.rt.TriggerUser(ctx, recipientID, "notification", map[string]any{
 			"kind": "admin_message",
 			"body": body,
 		}); err != nil {
@@ -745,7 +745,7 @@ func (s *NotificationService) BroadcastDiscipleJoined(ctx context.Context, actor
 			return nil
 		}
 		if s.rt != nil {
-			if err := s.rt.Trigger(ctx, "notifications."+uid, "notification", map[string]any{
+			if err := s.rt.TriggerUser(ctx, uid, "notification", map[string]any{
 				"kind":         NotifDiscipleJoined,
 				"actor_name":   name,
 				"actor_handle": actor.Handle,

pkg/realtime/auth.go

@@ -0,0 +1,119 @@
+package realtime
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+)
+
+// ErrNotConfigured is returned by the authorize helpers when realtime is
+// disabled. Callers should surface it as a 503: without Soketi there is no
+// subscription to authorize, and returning a bogus signature would just fail
+// opaquely at the client.
+var ErrNotConfigured = errors.New("realtime is not configured")
+
+// Channel-name prefixes, exported so consumers that parse channel names
+// (the Soketi webhook) stay in step with the ones that build them.
+const (
+	UserChannelPrefix       = "private-notifications."
+	LegacyUserChannelPrefix = "notifications."
+)
+
+// UserChannel is the per-user private channel carrying notification nudges.
+// Private (not public) so a subscription requires a signature the BFF only
+// issues to that user — without it, anyone who learns a user's UUID can read
+// their notification stream.
+func UserChannel(userID string) string { return UserChannelPrefix + userID }
+
+// LegacyUserChannel is the public channel the frontend used before private
+// channels existed. Still published to during the migration window so
+// service-worker-cached bundles keep working; delete once clients have
+// rolled over. See TriggerUser.
+func LegacyUserChannel(userID string) string { return LegacyUserChannelPrefix + userID }
+
+// AuthorizeChannel signs a private-channel subscription.
+//
+// Pusher's scheme: the client sends its socket_id and the channel it wants;
+// the app server — the only party holding the app secret — returns
+// "<key>:<hmac_sha256(socket_id:channel_name, secret)>". Soketi verifies with
+// its own copy of the secret. The signature is the whole authorization, so
+// callers MUST check the caller may access the channel before calling this.
+func (c *SoketiClient) AuthorizeChannel(socketID, channel string) (string, error) {
+	if c == nil {
+		return "", ErrNotConfigured
+	}
+	if socketID == "" || channel == "" {
+		return "", errors.New("socket_id and channel_name are required")
+	}
+	if strings.HasPrefix(channel, "presence-") {
+		return "", errors.New("presence channels must use AuthorizePresence")
+	}
+	return c.sign(socketID + ":" + channel), nil
+}
+
+// PresenceData is the member identity Soketi publishes to everyone else on a
+// presence channel. It is echoed to other subscribers verbatim, so UserInfo
+// must carry only what those members may already see — never an email.
+type PresenceData struct {
+	UserID   string         `json:"user_id"`
+	UserInfo map[string]any `json:"user_info,omitempty"`
+}
+
+// AuthorizePresence signs a presence-channel subscription, returning both the
+// auth signature and the channel_data the client must echo back. Presence
+// folds channel_data into the signed string, so the member identity can't be
+// tampered with client-side.
+func (c *SoketiClient) AuthorizePresence(socketID, channel string, data PresenceData) (auth, channelData string, err error) {
+	if c == nil {
+		return "", "", ErrNotConfigured
+	}
+	if socketID == "" || channel == "" {
+		return "", "", errors.New("socket_id and channel_name are required")
+	}
+	if data.UserID == "" {
+		return "", "", errors.New("presence data requires a user_id")
+	}
+	encoded, err := json.Marshal(data)
+	if err != nil {
+		return "", "", fmt.Errorf("marshal channel_data: %w", err)
+	}
+	channelData = string(encoded)
+	return c.sign(socketID + ":" + channel + ":" + channelData), channelData, nil
+}
+
+// sign returns the "<key>:<signature>" auth string for an already-built
+// string-to-sign.
+func (c *SoketiClient) sign(stringToSign string) string {
+	mac := hmac.New(sha256.New, []byte(c.secret))
+	mac.Write([]byte(stringToSign))
+	return c.key + ":" + hex.EncodeToString(mac.Sum(nil))
+}
+
+// TriggerUser publishes a per-user event to both the private channel and the
+// legacy public one.
+//
+// Dual-publish exists only for the migration: the frontend is a PWA, so a
+// service-worker-cached bundle can keep subscribing to the public channel for
+// a while after deploy, and a straight rename would silently stop its
+// notifications. Once clients have rolled over, drop the legacy Trigger and
+// this becomes a single publish.
+//
+// The private publish decides the error result; a legacy failure is not worth
+// failing the caller over.
+func (c *SoketiClient) TriggerUser(ctx context.Context, userID, event string, data any) error {
+	if c == nil {
+		return nil
+	}
+	err := c.Trigger(ctx, UserChannel(userID), event, data)
+	if legacyErr := c.Trigger(ctx, LegacyUserChannel(userID), event, data); legacyErr != nil {
+		c.logger.Debug("realtime: legacy channel publish failed", zap.Error(legacyErr))
+	}
+	return err
+}

pkg/realtime/auth_test.go

@@ -0,0 +1,93 @@
+package realtime
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"go.uber.org/zap"
+)
+
+// testClient builds a client with fixed credentials so signatures are
+// reproducible.
+func testClient(t *testing.T) *SoketiClient {
+	t.Helper()
+	c := NewSoketiClient("app-1", "test-key", "test-secret", "ws.example.com", 443, true, zap.NewNop())
+	if c == nil {
+		t.Fatal("NewSoketiClient returned nil for a fully configured client")
+	}
+	return c
+}
+
+// The expected signatures are HMAC-SHA256 vectors computed independently of
+// this package. Recomputing them with the same code under test would only
+// prove the code equals itself; a fixed vector catches a change to the
+// string-to-sign format, which is what Soketi actually verifies against.
+func TestAuthorizeChannel_MatchesKnownVector(t *testing.T) {
+	auth, err := testClient(t).AuthorizeChannel("1234.5678", "private-notifications.user-a")
+	if err != nil {
+		t.Fatalf("AuthorizeChannel: %v", err)
+	}
+	const want = "test-key:9c436d58a9d030f68b35d636bb3e930c471d5c11edec6745c969e4eee1c3e3df"
+	if auth != want {
+		t.Errorf("auth = %q, want %q", auth, want)
+	}
+}
+
+func TestAuthorizePresence_MatchesKnownVector(t *testing.T) {
+	auth, channelData, err := testClient(t).AuthorizePresence(
+		"1234.5678", "presence-conversation.c1", PresenceData{UserID: "user-a"})
+	if err != nil {
+		t.Fatalf("AuthorizePresence: %v", err)
+	}
+	if channelData != `{"user_id":"user-a"}` {
+		t.Errorf("channel_data = %q", channelData)
+	}
+	const want = "test-key:cf5a222b506d46383cb4edac7707ba620e4ba2bbb83f64d1cf6463b4241efd24"
+	if auth != want {
+		t.Errorf("auth = %q, want %q", auth, want)
+	}
+}
+
+// Presence signing folds channel_data into the signed string; routing a
+// presence channel through AuthorizeChannel would produce a signature Soketi
+// rejects, so it's refused outright rather than silently mis-signed.
+func TestAuthorizeChannel_RejectsPresenceChannel(t *testing.T) {
+	if _, err := testClient(t).AuthorizeChannel("1234.5678", "presence-conversation.c1"); err == nil {
+		t.Fatal("expected an error for a presence channel")
+	}
+}
+
+func TestAuthorize_NilClientReportsNotConfigured(t *testing.T) {
+	var c *SoketiClient
+	if _, err := c.AuthorizeChannel("1234.5678", "private-x"); !errors.Is(err, ErrNotConfigured) {
+		t.Errorf("AuthorizeChannel err = %v, want ErrNotConfigured", err)
+	}
+	if _, _, err := c.AuthorizePresence("1234.5678", "presence-x", PresenceData{UserID: "u"}); !errors.Is(err, ErrNotConfigured) {
+		t.Errorf("AuthorizePresence err = %v, want ErrNotConfigured", err)
+	}
+}
+
+func TestAuthorize_RequiresArguments(t *testing.T) {
+	c := testClient(t)
+	if _, err := c.AuthorizeChannel("", "private-x"); err == nil {
+		t.Error("expected an error for an empty socket_id")
+	}
+	if _, err := c.AuthorizeChannel("1234.5678", ""); err == nil {
+		t.Error("expected an error for an empty channel")
+	}
+	if _, _, err := c.AuthorizePresence("1234.5678", "presence-x", PresenceData{}); err == nil {
+		t.Error("expected an error for presence data with no user_id")
+	}
+}
+
+func TestUserChannels(t *testing.T) {
+	if got := UserChannel("u1"); got != "private-notifications.u1" {
+		t.Errorf("UserChannel = %q", got)
+	}
+	// The legacy name must stay public-prefixed: it exists so already-cached
+	// clients keep receiving events during the migration.
+	if got := LegacyUserChannel("u1"); got != "notifications.u1" || strings.HasPrefix(got, "private-") {
+		t.Errorf("LegacyUserChannel = %q", got)
+	}
+}