feat: add websocket

.env.example

@@ -57,4 +57,12 @@ FRONTEND_URL=http://localhost:3000
 GOOGLE_OAUTH_CLIENT_ID=
 GOOGLE_OAUTH_CLIENT_SECRET=
 FACEBOOK_OAUTH_CLIENT_ID=
-FACEBOOK_OAUTH_CLIENT_SECRET=
+FACEBOOK_OAUTH_CLIENT_SECRET=
+
+# SOKETI
+SOKETI_APP_ID=        
+SOKETI_APP_KEY=      
+SOKETI_APP_SECRET=
+SOKETI_HOST=ws.prabhupadasays.org   
+SOKETI_PORT=443   
+SOKETI_USE_TLS=true

internal/config/config.go

@@ -77,6 +77,18 @@ type Config struct {
 	GoogleOAuthClientSecret string
 	FacebookOAuthClientID   string
 	FacebookOAuthClientSecret string
+
+	// Soketi (Pusher-compatible WebSocket server) — used to push realtime
+	// notifications. Empty App credentials disable realtime delivery (the
+	// DB row + email still happen). SoketiHost is the HTTP-API origin the
+	// BFF triggers events against (e.g. "ws.prabhupadasays.org"); the
+	// frontend connects to the same host over WSS.
+	SoketiAppID     string
+	SoketiAppKey    string
+	SoketiAppSecret string
+	SoketiHost      string
+	SoketiPort      int
+	SoketiUseTLS    bool
 }
 
 // Load reads configuration from environment variables (with .env fallback).
@@ -135,6 +147,13 @@ func Load() (*Config, error) {
 		GoogleOAuthClientSecret:   getEnv("GOOGLE_OAUTH_CLIENT_SECRET", ""),
 		FacebookOAuthClientID:     getEnv("FACEBOOK_OAUTH_CLIENT_ID", ""),
 		FacebookOAuthClientSecret: getEnv("FACEBOOK_OAUTH_CLIENT_SECRET", ""),
+
+		SoketiAppID:     getEnv("SOKETI_APP_ID", ""),
+		SoketiAppKey:    getEnv("SOKETI_APP_KEY", ""),
+		SoketiAppSecret: getEnv("SOKETI_APP_SECRET", ""),
+		SoketiHost:      getEnv("SOKETI_HOST", "ws.prabhupadasays.org"),
+		SoketiPort:      getEnvInt("SOKETI_PORT", 443),
+		SoketiUseTLS:    getEnvBool("SOKETI_USE_TLS", true),
 	}
 
 	if cfg.MongoURI == "" {

internal/server/server.go

@@ -17,6 +17,7 @@ import (
 	"github.com/prabhupadasays/spsays-bff/pkg/auth"
 	"github.com/prabhupadasays/spsays-bff/pkg/mailer"
 	"github.com/prabhupadasays/spsays-bff/pkg/oauth"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
 	"github.com/prabhupadasays/spsays-bff/pkg/shorturl"
 	"go.uber.org/zap"
 )
@@ -197,7 +198,11 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 	blockRepo := repository.NewBlockRepo(s.pg)
 	reportRepo := repository.NewReportRepo(s.pg)
 	notifRepo := repository.NewNotificationRepo(s.pg)
-	notifSvc := service.NewNotificationService(notifRepo, userRepo, mail, logger, cfg.FrontendURL)
+	rtClient := realtime.NewSoketiClient(
+		cfg.SoketiAppID, cfg.SoketiAppKey, cfg.SoketiAppSecret,
+		cfg.SoketiHost, cfg.SoketiPort, cfg.SoketiUseTLS, logger,
+	)
+	notifSvc := service.NewNotificationService(notifRepo, userRepo, mail, rtClient, logger, cfg.FrontendURL)
 	graphSvc := service.NewSocialGraphService(followRepo, blockRepo, reportRepo, postRepo, userRepo, notifSvc, mail, cfg.FeedbackAdminEmails, logger)
 	s.relationH = handler.NewRelationshipHandler(graphSvc)
 	s.notificationH = handler.NewNotificationHandler(notifSvc)

internal/service/notification_service.go

@@ -21,6 +21,7 @@ import (
 	"github.com/prabhupadasays/spsays-bff/internal/model"
 	"github.com/prabhupadasays/spsays-bff/internal/repository"
 	"github.com/prabhupadasays/spsays-bff/pkg/mailer"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
 	"go.uber.org/zap"
 )
 
@@ -34,15 +35,17 @@ type NotificationService struct {
 	notifs      *repository.NotificationRepo
 	users       *repository.UserRepo
 	mail        mailer.Mailer
+	rt          *realtime.SoketiClient // optional — nil disables WebSocket push
 	logger      *zap.Logger
 	frontendURL string
 }
 
-func NewNotificationService(notifs *repository.NotificationRepo, users *repository.UserRepo, mail mailer.Mailer, logger *zap.Logger, frontendURL string) *NotificationService {
+func NewNotificationService(notifs *repository.NotificationRepo, users *repository.UserRepo, mail mailer.Mailer, rt *realtime.SoketiClient, logger *zap.Logger, frontendURL string) *NotificationService {
 	return &NotificationService{
 		notifs:      notifs,
 		users:       users,
 		mail:        mail,
+		rt:          rt,
 		logger:      logger,
 		frontendURL: strings.TrimRight(frontendURL, "/"),
 	}
@@ -92,10 +95,21 @@ func (s *NotificationService) dispatch(ctx context.Context, recipientID, actorID
 	if _, err := s.notifs.Create(ctx, recipientID, &actor, kind); err != nil {
 		s.logger.Warn("persist notification failed",
 			zap.String("recipient", recipientID), zap.String("kind", kind), zap.Error(err))
-		// Persistence failed, but still try to email — the email is the
-		// only delivery channel that exists today.
+		// Persistence failed, but still try the other channels.
 	}
 
+	// Look up the actor once — both the realtime push and the email use it.
+	actorUser, aerr := s.users.FindByID(ctx, actorID)
+	if aerr != nil {
+		s.logger.Warn("notification actor lookup failed", zap.String("actor", actorID), zap.Error(aerr))
+	}
+
+	// Realtime push (best-effort) — a lightweight nudge so the client can
+	// bump the bell + toast immediately. Fires regardless of whether the
+	// recipient has an email; authoritative data still comes from REST.
+	s.publishRealtime(ctx, recipientID, kind, actorUser)
+
+	// Email (best-effort).
 	recipient, err := s.users.FindByID(ctx, recipientID)
 	if err != nil {
 		s.logger.Warn("notification recipient lookup failed", zap.String("recipient", recipientID), zap.Error(err))
@@ -103,12 +117,11 @@ func (s *NotificationService) dispatch(ctx context.Context, recipientID, actorID
 	}
 	if recipient.Email == nil || strings.TrimSpace(*recipient.Email) == "" {
 		// No email on file (OAuth-only account that never added one) —
-		// the persisted row is the notification; nothing to send.
+		// the persisted row + realtime push are the notification.
 		return
 	}
-	actorUser, err := s.users.FindByID(ctx, actorID)
-	if err != nil {
-		s.logger.Warn("notification actor lookup failed", zap.String("actor", actorID), zap.Error(err))
+	if actorUser == nil {
+		// Can't render the email body without the actor's name.
 		return
 	}
 
@@ -124,6 +137,27 @@ func (s *NotificationService) dispatch(ctx context.Context, recipientID, actorID
 	}
 }
 
+// publishRealtime pushes a lightweight notification event to the
+// recipient's Soketi channel. PUBLIC channel for now ("notifications.{id}")
+// — switch to a "private-" prefix once Soketi App auth is wired up. The
+// payload carries just enough for a toast; the client refetches the
+// authoritative unread count + list from REST.
+func (s *NotificationService) publishRealtime(ctx context.Context, recipientID, kind string, actor *model.User) {
+	if s.rt == nil {
+		return
+	}
+	payload := map[string]any{"kind": kind}
+	if actor != nil {
+		payload["actor_name"] = actorDisplay(actor)
+		payload["actor_handle"] = actor.Handle
+		payload["actor_avatar_url"] = actor.AvatarURL
+	}
+	if err := s.rt.Trigger(ctx, "notifications."+recipientID, "notification", payload); err != nil {
+		s.logger.Warn("realtime notification publish failed",
+			zap.String("recipient", recipientID), zap.String("kind", kind), zap.Error(err))
+	}
+}
+
 func (s *NotificationService) render(kind string, recipient, actor *model.User) (string, error) {
 	data := map[string]any{
 		"RecipientName": recipient.Name,

pkg/realtime/soketi.go

@@ -0,0 +1,144 @@
+// Package realtime publishes events to a Soketi (Pusher-compatible)
+// WebSocket server so the frontend can receive realtime notifications.
+//
+// We implement the Pusher REST "trigger" protocol directly (HMAC-SHA256
+// signed POST to /apps/{id}/events) rather than pull in a client library —
+// it's a small, stable wire format and avoids a new dependency.
+//
+// Channel auth is intentionally NOT handled here: we publish to PUBLIC
+// channels for now (see the notification service). When Soketi Apps are
+// configured for connection/channel authentication, switch the channel
+// names to the `private-` prefix and add a /broadcasting/auth endpoint —
+// the trigger code below is unchanged by that.
+package realtime
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/md5"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// SoketiClient triggers events on a Soketi/Pusher app. A nil *SoketiClient
+// is valid and makes Trigger a no-op, so callers can hold an optional
+// client without nil-checks.
+type SoketiClient struct {
+	appID   string
+	key     string
+	secret  string
+	baseURL string // scheme://host[:port]
+	http    *http.Client
+	logger  *zap.Logger
+}
+
+// NewSoketiClient returns a configured client, or nil when any required
+// credential is missing (realtime disabled — callers degrade gracefully).
+func NewSoketiClient(appID, key, secret, host string, port int, useTLS bool, logger *zap.Logger) *SoketiClient {
+	if appID == "" || key == "" || secret == "" || host == "" {
+		logger.Info("realtime: Soketi not configured — notifications won't be pushed over WebSocket")
+		return nil
+	}
+	scheme := "http"
+	if useTLS {
+		scheme = "https"
+	}
+	hostport := host
+	// Omit the port when it's the scheme default (cleaner, and matches
+	// what most reverse-proxied setups expect).
+	if port != 0 && !((useTLS && port == 443) || (!useTLS && port == 80)) {
+		hostport = fmt.Sprintf("%s:%d", host, port)
+	}
+	return &SoketiClient{
+		appID:   appID,
+		key:     key,
+		secret:  secret,
+		baseURL: scheme + "://" + hostport,
+		http:    &http.Client{Timeout: 5 * time.Second},
+		logger:  logger,
+	}
+}
+
+// Trigger publishes `event` with `data` (JSON-marshalled) to `channel`.
+// Best-effort: on a nil client it no-ops; errors are returned for the
+// caller to log-and-ignore.
+func (c *SoketiClient) Trigger(ctx context.Context, channel, event string, data any) error {
+	if c == nil {
+		return nil
+	}
+
+	dataJSON, err := json.Marshal(data)
+	if err != nil {
+		return fmt.Errorf("marshal event data: %w", err)
+	}
+	// Pusher wraps the payload as a JSON STRING in the `data` field.
+	body, err := json.Marshal(map[string]any{
+		"name":     event,
+		"channels": []string{channel},
+		"data":     string(dataJSON),
+	})
+	if err != nil {
+		return fmt.Errorf("marshal trigger body: %w", err)
+	}
+
+	bodyMD5 := fmt.Sprintf("%x", md5.Sum(body))
+	path := "/apps/" + c.appID + "/events"
+
+	// Auth params (all values are signature-safe, no URL-encoding needed),
+	// sorted alphabetically by key for the canonical string-to-sign.
+	params := map[string]string{
+		"auth_key":       c.key,
+		"auth_timestamp": strconv.FormatInt(time.Now().Unix(), 10),
+		"auth_version":   "1.0",
+		"body_md5":       bodyMD5,
+	}
+	keys := make([]string, 0, len(params))
+	for k := range params {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	var qs strings.Builder
+	for i, k := range keys {
+		if i > 0 {
+			qs.WriteByte('&')
+		}
+		qs.WriteString(k)
+		qs.WriteByte('=')
+		qs.WriteString(params[k])
+	}
+	query := qs.String()
+
+	stringToSign := "POST\n" + path + "\n" + query
+	mac := hmac.New(sha256.New, []byte(c.secret))
+	mac.Write([]byte(stringToSign))
+	signature := hex.EncodeToString(mac.Sum(nil))
+
+	fullURL := c.baseURL + path + "?" + query + "&auth_signature=" + signature
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, fullURL, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("build trigger request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("trigger request: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		b, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return fmt.Errorf("soketi trigger %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
+	}
+	return nil
+}