From e352948fb1d9012098593fbf491fedc04570653d Mon Sep 17 00:00:00 2001 From: Krispire Admin Date: Sun, 7 Jun 2026 00:23:21 +0530 Subject: [PATCH] feat: add websocket --- .env.example | 10 +- internal/config/config.go | 19 +++ internal/server/server.go | 7 +- internal/service/notification_service.go | 48 ++++++-- pkg/realtime/soketi.go | 144 +++++++++++++++++++++++ 5 files changed, 219 insertions(+), 9 deletions(-) create mode 100644 pkg/realtime/soketi.go diff --git a/.env.example b/.env.example index b32c627..23d9e09 100644 --- a/.env.example +++ b/.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= \ No newline at end of file +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 diff --git a/internal/config/config.go b/internal/config/config.go index bacc10b..4786720 100644 --- a/internal/config/config.go +++ b/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 == "" { diff --git a/internal/server/server.go b/internal/server/server.go index 375dd27..0add573 100644 --- a/internal/server/server.go +++ b/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) diff --git a/internal/service/notification_service.go b/internal/service/notification_service.go index fdaab1c..57a2a58 100644 --- a/internal/service/notification_service.go +++ b/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, diff --git a/pkg/realtime/soketi.go b/pkg/realtime/soketi.go new file mode 100644 index 0000000..bd8cf67 --- /dev/null +++ b/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 +}