diff --git a/internal/handler/broadcast.go b/internal/handler/broadcast.go new file mode 100644 index 0000000..c3953f1 --- /dev/null +++ b/internal/handler/broadcast.go @@ -0,0 +1,151 @@ +package handler + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + "strings" + "time" + + "github.com/go-chi/chi/v5" + "github.com/prabhupadasays/spsays-bff/internal/middleware" + "github.com/prabhupadasays/spsays-bff/internal/repository" + "github.com/prabhupadasays/spsays-bff/internal/service" + "github.com/prabhupadasays/spsays-bff/pkg/response" +) + +// BroadcastHandler serves both sides of announcements: the admin routes that +// compose and publish them, and the two app routes that fetch and dismiss. +type BroadcastHandler struct { + svc *service.BroadcastService +} + +func NewBroadcastHandler(svc *service.BroadcastService) *BroadcastHandler { + return &BroadcastHandler{svc: svc} +} + +func writeBroadcastError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, repository.ErrBroadcastGone): + response.Error(w, http.StatusNotFound, "broadcast not found") + case errors.Is(err, service.ErrBroadcastTitle), + errors.Is(err, service.ErrBroadcastBody), + errors.Is(err, service.ErrBroadcastCTA), + errors.Is(err, service.ErrBroadcastBanner): + response.Error(w, http.StatusBadRequest, err.Error()) + default: + response.Error(w, http.StatusInternalServerError, "broadcast request failed") + } +} + +// ── App routes ──────────────────────────────────────────────────────────── + +// Pending — GET /broadcasts/pending. Returns null when nothing is waiting, so +// the client can call it on load and after a nudge without special-casing. +func (h *BroadcastHandler) Pending(w http.ResponseWriter, r *http.Request) { + b, err := h.svc.Pending(r.Context(), middleware.UserID(r.Context())) + if err != nil { + writeBroadcastError(w, err) + return + } + response.JSON(w, http.StatusOK, b) +} + +// Seen — POST /broadcasts/{id}/seen. The "at most once" half; idempotent. +func (h *BroadcastHandler) Seen(w http.ResponseWriter, r *http.Request) { + if err := h.svc.MarkSeen(r.Context(), chi.URLParam(r, "id"), + middleware.UserID(r.Context())); err != nil { + writeBroadcastError(w, err) + return + } + response.NoContent(w) +} + +// ── Admin routes ────────────────────────────────────────────────────────── + +type createBroadcastReq struct { + Title string `json:"title"` + Body string `json:"body"` + BannerKey *string `json:"banner_key"` + CTALabel *string `json:"cta_label"` + CTAURL *string `json:"cta_url"` + // ExpiresAt stops the announcement being offered. Without one, someone + // signing up months later would still be shown it. + ExpiresAt *time.Time `json:"expires_at"` + // Publish immediately rather than saving a draft. + Publish bool `json:"publish"` +} + +// Create — POST /admin/broadcasts +func (h *BroadcastHandler) Create(w http.ResponseWriter, r *http.Request) { + var req createBroadcastReq + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + response.Error(w, http.StatusBadRequest, "invalid JSON body") + return + } + b, err := h.svc.Create(r.Context(), middleware.UserID(r.Context()), service.CreateParams{ + Title: req.Title, Body: req.Body, BannerKey: req.BannerKey, + CTALabel: req.CTALabel, CTAURL: req.CTAURL, ExpiresAt: req.ExpiresAt, + }) + if err != nil { + writeBroadcastError(w, err) + return + } + if req.Publish { + published, err := h.svc.Publish(r.Context(), b.ID) + if err != nil { + writeBroadcastError(w, err) + return + } + b = published + } + response.JSON(w, http.StatusCreated, b) +} + +// Publish — POST /admin/broadcasts/{id}/publish +func (h *BroadcastHandler) Publish(w http.ResponseWriter, r *http.Request) { + b, err := h.svc.Publish(r.Context(), chi.URLParam(r, "id")) + if err != nil { + writeBroadcastError(w, err) + return + } + response.JSON(w, http.StatusOK, b) +} + +// List — GET /admin/broadcasts +func (h *BroadcastHandler) List(w http.ResponseWriter, r *http.Request) { + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + items, err := h.svc.List(r.Context(), limit) + if err != nil { + writeBroadcastError(w, err) + return + } + response.JSON(w, http.StatusOK, items) +} + +// Delete — DELETE /admin/broadcasts/{id} +func (h *BroadcastHandler) Delete(w http.ResponseWriter, r *http.Request) { + if err := h.svc.Delete(r.Context(), chi.URLParam(r, "id")); err != nil { + writeBroadcastError(w, err) + return + } + response.NoContent(w) +} + +// PresignBanner — POST /admin/broadcasts/banner-presign +func (h *BroadcastHandler) PresignBanner(w http.ResponseWriter, r *http.Request) { + var req struct { + ContentType string `json:"content_type"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + response.Error(w, http.StatusBadRequest, "invalid JSON body") + return + } + upload, key, err := h.svc.PresignBanner(r.Context(), strings.TrimSpace(req.ContentType)) + if err != nil { + response.Error(w, http.StatusBadRequest, err.Error()) + return + } + response.JSON(w, http.StatusOK, map[string]string{"upload_url": upload, "key": key}) +} diff --git a/internal/model/broadcast.go b/internal/model/broadcast.go new file mode 100644 index 0000000..c0dcfab --- /dev/null +++ b/internal/model/broadcast.go @@ -0,0 +1,40 @@ +package model + +import ( + "encoding/json" + "time" +) + +// Broadcast is an announcement shown to every user exactly once. +// +// Deliberately not a notification: it creates no tray rows and no per-user +// records until someone actually dismisses it. The modal is the delivery +// surface; push and the realtime nudge only tell clients to go look. +type Broadcast struct { + ID string `json:"id"` + Title string `json:"title"` + Body string `json:"body"` + // Entities are offset/length formatting spans, the same shape chat uses, + // so no HTML is ever stored or rendered. + Entities json.RawMessage `json:"entities"` + // BannerURL is resolved from the stored R2 key for the client. + BannerURL *string `json:"banner_url,omitempty"` + CTALabel *string `json:"cta_label,omitempty"` + CTAURL *string `json:"cta_url,omitempty"` + + PublishedAt *time.Time `json:"published_at,omitempty"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + + // SeenCount is filled for the admin list only — how many people have + // dismissed it so far. + SeenCount int64 `json:"seen_count,omitempty"` +} + +// Live reports whether the broadcast is currently deliverable. +func (b *Broadcast) Live(now time.Time) bool { + if b.PublishedAt == nil || b.PublishedAt.After(now) { + return false + } + return b.ExpiresAt == nil || b.ExpiresAt.After(now) +} diff --git a/internal/repository/broadcast_repo.go b/internal/repository/broadcast_repo.go new file mode 100644 index 0000000..1b68291 --- /dev/null +++ b/internal/repository/broadcast_repo.go @@ -0,0 +1,151 @@ +package repository + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/prabhupadasays/spsays-bff/internal/model" +) + +var ErrBroadcastGone = errors.New("broadcast not found") + +type BroadcastRepo struct{ pg *Postgres } + +func NewBroadcastRepo(pg *Postgres) *BroadcastRepo { return &BroadcastRepo{pg: pg} } + +const broadcastColumns = `id, title, body, entities, banner_key, cta_label, + cta_url, published_at, expires_at, created_at` + +// scanBroadcast decodes a row produced by broadcastColumns. bannerKey stays +// out of the model — the service turns it into a URL. +func scanBroadcast(row pgx.Row, b *model.Broadcast, bannerKey **string) error { + return row.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, bannerKey, + &b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt) +} + +type CreateBroadcastParams struct { + Title string + Body string + Entities []byte + BannerKey *string + CTALabel *string + CTAURL *string + ExpiresAt *time.Time + CreatedBy string +} + +func (r *BroadcastRepo) Create(ctx context.Context, p CreateBroadcastParams) (*model.Broadcast, *string, error) { + var b model.Broadcast + var key *string + err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` + INSERT INTO broadcasts (title, body, entities, banner_key, cta_label, + cta_url, expires_at, created_by) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8) + RETURNING `+broadcastColumns, + p.Title, p.Body, p.Entities, p.BannerKey, p.CTALabel, p.CTAURL, + p.ExpiresAt, p.CreatedBy), &b, &key) + if err != nil { + return nil, nil, fmt.Errorf("create broadcast: %w", err) + } + return &b, key, nil +} + +// Publish stamps published_at, making the broadcast deliverable. Idempotent: +// re-publishing keeps the original timestamp so the live window doesn't shift. +func (r *BroadcastRepo) Publish(ctx context.Context, id string) (*model.Broadcast, *string, error) { + var b model.Broadcast + var key *string + err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` + UPDATE broadcasts + SET published_at = COALESCE(published_at, now()), updated_at = now() + WHERE id = $1 + RETURNING `+broadcastColumns, id), &b, &key) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil, ErrBroadcastGone + } + if err != nil { + return nil, nil, fmt.Errorf("publish broadcast: %w", err) + } + return &b, key, nil +} + +func (r *BroadcastRepo) Delete(ctx context.Context, id string) error { + tag, err := r.pg.Pool.Exec(ctx, `DELETE FROM broadcasts WHERE id = $1`, id) + if err != nil { + return fmt.Errorf("delete broadcast: %w", err) + } + if tag.RowsAffected() == 0 { + return ErrBroadcastGone + } + return nil +} + +// Pending returns the newest live broadcast this user has not dismissed. +func (r *BroadcastRepo) Pending(ctx context.Context, userID string) (*model.Broadcast, *string, error) { + var b model.Broadcast + var key *string + err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` + SELECT `+broadcastColumns+` + FROM broadcasts b + WHERE b.published_at IS NOT NULL + AND b.published_at <= now() + AND (b.expires_at IS NULL OR b.expires_at > now()) + AND NOT EXISTS ( + SELECT 1 FROM broadcast_views v + WHERE v.broadcast_id = b.id AND v.user_id = $1) + -- Newest first. Announcements are time-sensitive, so a stale one + -- must not gate a fresh one; dismissing this reveals the next, so + -- nothing is skipped either way — only the order differs. + ORDER BY b.published_at DESC + LIMIT 1`, userID), &b, &key) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil, nil // nothing waiting is not an error + } + if err != nil { + return nil, nil, fmt.Errorf("pending broadcast: %w", err) + } + return &b, key, nil +} + +// MarkSeen records a dismissal. Idempotent — two tabs can race. +func (r *BroadcastRepo) MarkSeen(ctx context.Context, id, userID string) error { + _, err := r.pg.Pool.Exec(ctx, ` + INSERT INTO broadcast_views (broadcast_id, user_id) VALUES ($1,$2) + ON CONFLICT DO NOTHING`, id, userID) + if err != nil { + return fmt.Errorf("mark broadcast seen: %w", err) + } + return nil +} + +// List returns broadcasts newest first, with dismissal counts, for the admin. +func (r *BroadcastRepo) List(ctx context.Context, limit int) ([]model.Broadcast, []*string, error) { + rows, err := r.pg.Pool.Query(ctx, ` + SELECT `+broadcastColumns+`, + (SELECT count(*) FROM broadcast_views v WHERE v.broadcast_id = b.id) + FROM broadcasts b + ORDER BY COALESCE(b.published_at, b.created_at) DESC + LIMIT $1`, limit) + if err != nil { + return nil, nil, fmt.Errorf("list broadcasts: %w", err) + } + defer rows.Close() + + out := []model.Broadcast{} + keys := []*string{} + for rows.Next() { + var b model.Broadcast + var key *string + if err := rows.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, &key, + &b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt, + &b.SeenCount); err != nil { + return nil, nil, fmt.Errorf("scan broadcast: %w", err) + } + out = append(out, b) + keys = append(keys, key) + } + return out, keys, rows.Err() +} diff --git a/internal/repository/push_repo.go b/internal/repository/push_repo.go index 22119b9..ac09a72 100644 --- a/internal/repository/push_repo.go +++ b/internal/repository/push_repo.go @@ -54,6 +54,30 @@ func (r *PushRepo) ListByUser(ctx context.Context, userID string) ([]model.PushS // PruneToMostRecent keeps only the `keep` most-recently-used subscriptions for // a user, deleting older ones — a soft per-user device cap. +// SubscribedUserIDs lists everyone with at least one live push subscription. +// +// Used by broadcasts, which fan out to the whole user base: iterating users +// and asking each for subscriptions would query for the (large) majority who +// have never enabled push. +func (r *PushRepo) SubscribedUserIDs(ctx context.Context) ([]string, error) { + rows, err := r.pg.Pool.Query(ctx, + `SELECT DISTINCT user_id FROM push_subscriptions`) + if err != nil { + return nil, fmt.Errorf("subscribed users: %w", err) + } + defer rows.Close() + + out := []string{} + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan subscribed user: %w", err) + } + out = append(out, id) + } + return out, rows.Err() +} + func (r *PushRepo) PruneToMostRecent(ctx context.Context, userID string, keep int) error { if keep < 1 { keep = 1 diff --git a/internal/server/server.go b/internal/server/server.go index 3b5e055..d5896c4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -77,6 +77,7 @@ type Server struct { feedbackH *handler.FeedbackHandler statsH *handler.StatsHandler adminH *handler.AdminHandler + broadcastH *handler.BroadcastHandler // Auth state (nil when Postgres is unavailable — /auth and /users return 503) authIssuer *auth.Issuer @@ -343,6 +344,14 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { adminRepo := repository.NewAdminRepo(s.pg) adminSvc := service.NewAdminService(adminRepo, statsRepo, notifSvc, feedbackSvc, userRepo, s.discipleRepo, mail, logger, cfg.FrontendURL) s.adminH = handler.NewAdminHandler(adminSvc) + + // Broadcasts — admin announcements shown once per user. No notification + // rows: the modal is the surface, and push plus the realtime nudge only + // tell clients to go and check for one. + broadcastSvc := service.NewBroadcastService( + repository.NewBroadcastRepo(s.pg), pushSender, pushRepo, + notifPrefsRepo, rtClient, r2, logger) + s.broadcastH = handler.NewBroadcastHandler(broadcastSvc) } // initShortener wires Postgres + Redis + the shorturl handler/workers. @@ -912,6 +921,15 @@ func (s *Server) setupRouter() { // Chat — inbox, threads, messages. Every route requires a session; // membership is re-checked per request in the service, so a // conversation id on its own grants nothing. + if s.broadcastH != nil { + // Announcements: fetch the one waiting, and dismiss it. + r.Route("/broadcasts", func(r chi.Router) { + r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo)) + r.Get("/pending", s.broadcastH.Pending) + r.Post("/{id}/seen", s.broadcastH.Seen) + }) + } + if s.chatH != nil { r.Route("/chat", func(r chi.Router) { r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo)) @@ -1133,6 +1151,13 @@ func (s *Server) setupRouter() { // Direct-disciple linking: mark an account as one of Śrīla // Prabhupāda's initiated disciples (or clear the mark). r.Get("/disciples", s.adminH.ListDisciples) + + // Announcements composed and published from the admin panel. + r.Get("/broadcasts", s.broadcastH.List) + r.Post("/broadcasts", s.broadcastH.Create) + r.Post("/broadcasts/banner-presign", s.broadcastH.PresignBanner) + r.Post("/broadcasts/{id}/publish", s.broadcastH.Publish) + r.Delete("/broadcasts/{id}", s.broadcastH.Delete) r.Post("/users/{id}/disciple", s.adminH.LinkDisciple) r.Delete("/users/{id}/disciple", s.adminH.UnlinkDisciple) r.Get("/posts", s.adminH.ListPosts) diff --git a/internal/service/broadcast_service.go b/internal/service/broadcast_service.go new file mode 100644 index 0000000..7c1b4c7 --- /dev/null +++ b/internal/service/broadcast_service.go @@ -0,0 +1,266 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/url" + "strings" + "time" + + "github.com/prabhupadasays/spsays-bff/internal/model" + "github.com/prabhupadasays/spsays-bff/internal/repository" + "github.com/prabhupadasays/spsays-bff/pkg/realtime" + "go.uber.org/zap" +) + +var ( + ErrBroadcastTitle = errors.New("a title is required") + ErrBroadcastBody = errors.New("a description is required") + ErrBroadcastCTA = errors.New("that CTA link is not a valid URL") + ErrBroadcastBanner = errors.New("that image was not uploaded for a broadcast") +) + +const ( + maxBroadcastTitleRunes = 200 + maxBroadcastBodyRunes = 4000 + // bannerPrefix namespaces broadcast banners in the PUBLIC bucket. An + // announcement is not private, and a signed URL would expire while the + // modal sat open on someone's screen. + bannerPrefix = "broadcasts/" +) + +// BroadcastChannel is the PUBLIC realtime channel every client watches for +// announcements. +// +// Public is deliberate and safe because the event carries no content — just +// "something was published". Each client then fetches /broadcasts/pending, +// which is authenticated and per-user, so someone who has already dismissed it +// gets nothing. That also keeps publishing to one Soketi call instead of one +// per user. +const BroadcastChannel = "broadcasts" + +type BroadcastService struct { + repo *repository.BroadcastRepo + push *PushSender + pushes *repository.PushRepo + prefs *repository.NotificationPrefsRepo + rt *realtime.SoketiClient + r2 *repository.R2 + logger *zap.Logger +} + +func NewBroadcastService( + repo *repository.BroadcastRepo, + push *PushSender, + pushes *repository.PushRepo, + prefs *repository.NotificationPrefsRepo, + rt *realtime.SoketiClient, + r2 *repository.R2, + logger *zap.Logger, +) *BroadcastService { + return &BroadcastService{ + repo: repo, push: push, pushes: pushes, prefs: prefs, + rt: rt, r2: r2, logger: logger, + } +} + +// CreateParams is an admin-authored announcement. +type CreateParams struct { + Title string + Body string + BannerKey *string + CTALabel *string + CTAURL *string + ExpiresAt *time.Time +} + +// Create stores a draft. Nothing is delivered until Publish. +func (s *BroadcastService) Create(ctx context.Context, adminID string, p CreateParams) (*model.Broadcast, error) { + title := strings.TrimSpace(p.Title) + body := strings.TrimSpace(p.Body) + if title == "" || len([]rune(title)) > maxBroadcastTitleRunes { + return nil, ErrBroadcastTitle + } + if body == "" || len([]rune(body)) > maxBroadcastBodyRunes { + return nil, ErrBroadcastBody + } + if p.CTAURL != nil && strings.TrimSpace(*p.CTAURL) != "" { + clean, ok := safeCTAURL(*p.CTAURL) + if !ok { + return nil, ErrBroadcastCTA + } + p.CTAURL = &clean + } else { + p.CTAURL = nil + } + if p.BannerKey != nil && !strings.HasPrefix(*p.BannerKey, bannerPrefix) { + return nil, ErrBroadcastBanner + } + + // Same parser as the chat composer, so the admin writes *bold* / ~strike~ / + // `code` and links are detected — and no HTML is ever stored. + clean, entities := ParseEntities(body) + encoded, err := json.Marshal(entities) + if err != nil { + return nil, fmt.Errorf("encode entities: %w", err) + } + if entities == nil { + encoded = []byte(`[]`) + } + + b, key, err := s.repo.Create(ctx, repository.CreateBroadcastParams{ + Title: title, Body: clean, Entities: encoded, BannerKey: p.BannerKey, + CTALabel: p.CTALabel, CTAURL: p.CTAURL, ExpiresAt: p.ExpiresAt, + CreatedBy: adminID, + }) + if err != nil { + return nil, err + } + s.withBanner(b, key) + return b, nil +} + +// Publish makes a broadcast live, then nudges everyone: connected clients over +// realtime, absent ones by push. +func (s *BroadcastService) Publish(ctx context.Context, id string) (*model.Broadcast, error) { + b, key, err := s.repo.Publish(ctx, id) + if err != nil { + return nil, err + } + s.withBanner(b, key) + + // One trigger for the whole user base. Content-free on purpose — clients + // re-fetch their own pending state. + if s.rt != nil { + if err := s.rt.Trigger(ctx, BroadcastChannel, "broadcast_published", map[string]any{}); err != nil { + s.logger.Warn("broadcast: realtime nudge failed", zap.Error(err)) + } + } + go s.fanOutPush(context.WithoutCancel(ctx), b) + return b, nil +} + +// fanOutPush notifies everyone with a push subscription who hasn't opted out +// of announcements. +// +// Runs detached from the request: a large user base takes far longer than an +// admin should wait on a button. +func (s *BroadcastService) fanOutPush(ctx context.Context, b *model.Broadcast) { + if s.push == nil || !s.push.Enabled() || s.pushes == nil { + return + } + users, err := s.pushes.SubscribedUserIDs(ctx) + if err != nil { + s.logger.Warn("broadcast: listing subscribers failed", zap.Error(err)) + return + } + + sent := 0 + for _, uid := range users { + if s.prefs != nil { + // Reuses the existing "Community" category, described in settings + // as rare announcements — which is exactly this. Someone who + // opted out of those stays opted out. + if off, err := s.prefs.DisabledSet(ctx, uid); err == nil && + prefOff(off, NotifDiscipleJoined, repository.ChannelPush) { + continue + } + } + s.push.SendToUser(ctx, uid, PushPayload{ + Title: b.Title, + Body: truncateRunes(b.Body, 160), + URL: "/", + // One tag per broadcast: a re-publish replaces rather than stacks. + Tag: "broadcast:" + b.ID, + }) + sent++ + } + s.logger.Info("broadcast: push fan-out complete", + zap.String("broadcast_id", b.ID), zap.Int("recipients", sent)) +} + +// Pending returns the announcement waiting for this user, or nil. +func (s *BroadcastService) Pending(ctx context.Context, userID string) (*model.Broadcast, error) { + b, key, err := s.repo.Pending(ctx, userID) + if err != nil || b == nil { + return nil, err + } + s.withBanner(b, key) + return b, nil +} + +// MarkSeen records that a user dismissed it — the "at most once" half. +func (s *BroadcastService) MarkSeen(ctx context.Context, id, userID string) error { + return s.repo.MarkSeen(ctx, id, userID) +} + +func (s *BroadcastService) List(ctx context.Context, limit int) ([]model.Broadcast, error) { + if limit <= 0 || limit > 100 { + limit = 50 + } + items, keys, err := s.repo.List(ctx, limit) + if err != nil { + return nil, err + } + for i := range items { + s.withBanner(&items[i], keys[i]) + } + return items, nil +} + +func (s *BroadcastService) Delete(ctx context.Context, id string) error { + return s.repo.Delete(ctx, id) +} + +// PresignBanner issues an upload URL for a banner image. +func (s *BroadcastService) PresignBanner(ctx context.Context, contentType string) (uploadURL, key string, err error) { + if s.r2 == nil { + return "", "", errors.New("media uploads are not configured") + } + kind, ok := attachmentKindFor(contentType) + if !ok || kind != model.AttachmentImage { + return "", "", ErrChatBadMedia + } + return presignImage(ctx, s.r2, strings.TrimSuffix(bannerPrefix, "/"), "banner", contentType) +} + +func (s *BroadcastService) withBanner(b *model.Broadcast, key *string) { + if key == nil || *key == "" || s.r2 == nil { + return + } + url := s.r2.PublicURL(*key) + b.BannerURL = &url +} + +// safeCTAURL accepts an in-app path or an absolute http(s) URL. +// +// Admin-authored, so this is a typo guard more than a threat model — but +// `javascript:` and `data:` still have no business in a link the whole user +// base is invited to tap. +func safeCTAURL(raw string) (string, bool) { + raw = strings.TrimSpace(raw) + if raw == "" || len(raw) > 2048 { + return "", false + } + if strings.ContainsAny(raw, "\\") || strings.ContainsAny(raw, "\n\r\t") { + return "", false + } + if strings.HasPrefix(raw, "/") { + // Protocol-relative "//evil.example" is not an in-app path. + if strings.HasPrefix(raw, "//") { + return "", false + } + u, err := url.Parse(raw) + if err != nil || u.Scheme != "" || u.Host != "" { + return "", false + } + return raw, true + } + u, err := url.Parse(raw) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { + return "", false + } + return u.String(), true +} diff --git a/internal/service/broadcast_service_test.go b/internal/service/broadcast_service_test.go new file mode 100644 index 0000000..e7cd1b6 --- /dev/null +++ b/internal/service/broadcast_service_test.go @@ -0,0 +1,166 @@ +package service + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/google/uuid" + "github.com/joho/godotenv" + "github.com/prabhupadasays/spsays-bff/internal/repository" + "go.uber.org/zap" +) + +func testBroadcastService(t *testing.T) (*BroadcastService, *repository.Postgres) { + t.Helper() + uri := os.Getenv("POSTGRES_URI") + if uri == "" { + _ = godotenv.Load(filepath.Join("..", "..", ".env")) + uri = os.Getenv("POSTGRES_URI") + } + if uri == "" { + t.Skip("POSTGRES_URI not set") + } + pg, err := repository.NewPostgres(context.Background(), uri, zap.NewNop()) + if err != nil { + t.Skipf("postgres unreachable: %v", err) + } + t.Cleanup(pg.Close) + // Nil push/realtime/R2: these tests are about who sees what, and none of + // those services are reachable from a test run. + return NewBroadcastService(repository.NewBroadcastRepo(pg), nil, nil, nil, nil, nil, zap.NewNop()), pg +} + +func bcUser(t *testing.T, pg *repository.Postgres, label string) string { + t.Helper() + id := uuid.NewString() + suffix := id[:8] + if _, err := pg.Pool.Exec(context.Background(), ` + INSERT INTO users (id, email, password_hash, name, handle) + VALUES ($1, $2, 'x', $3, $4)`, + id, "bc-"+suffix+"@example.invalid", "BC "+label, "bc_"+suffix); err != nil { + t.Fatalf("create user: %v", err) + } + t.Cleanup(func() { + _, _ = pg.Pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, id) + }) + return id +} + +func cleanupBroadcast(t *testing.T, pg *repository.Postgres, id string) { + t.Cleanup(func() { + _, _ = pg.Pool.Exec(context.Background(), `DELETE FROM broadcasts WHERE id = $1`, id) + }) +} + +// "At least and at most once" is the whole contract. A user must be offered a +// published announcement until they dismiss it, and never again after. +func TestBroadcast_ShownExactlyOncePerUser(t *testing.T) { + svc, pg := testBroadcastService(t) + ctx := context.Background() + admin, alice, bob := bcUser(t, pg, "Admin"), bcUser(t, pg, "A"), bcUser(t, pg, "B") + + b, err := svc.Create(ctx, admin, CreateParams{ + Title: "New: chat", Body: "Direct messages are here.", + }) + if err != nil { + t.Fatal(err) + } + cleanupBroadcast(t, pg, b.ID) + + // A draft is nobody's business yet. + if got, _ := svc.Pending(ctx, alice); got != nil { + t.Fatalf("an unpublished draft was offered: %+v", got) + } + + if _, err := svc.Publish(ctx, b.ID); err != nil { + t.Fatal(err) + } + + // At least once: offered, and still offered until dismissed. + for i := 0; i < 2; i++ { + got, err := svc.Pending(ctx, alice) + if err != nil { + t.Fatal(err) + } + if got == nil || got.ID != b.ID { + t.Fatalf("call %d: expected the published announcement, got %+v", i+1, got) + } + } + + // At most once: dismissed means gone, and dismissing twice is harmless. + if err := svc.MarkSeen(ctx, b.ID, alice); err != nil { + t.Fatal(err) + } + if err := svc.MarkSeen(ctx, b.ID, alice); err != nil { + t.Errorf("second dismissal should be idempotent, got %v", err) + } + if got, _ := svc.Pending(ctx, alice); got != nil { + t.Errorf("shown again after dismissal: %+v", got) + } + + // One person dismissing must not dismiss it for everyone. + if got, _ := svc.Pending(ctx, bob); got == nil || got.ID != b.ID { + t.Errorf("another user lost the announcement: %+v", got) + } +} + +// An announcement past its expiry stops being offered, so someone signing up +// months later isn't shown last quarter's news. +func TestBroadcast_ExpiryStopsDelivery(t *testing.T) { + svc, pg := testBroadcastService(t) + ctx := context.Background() + admin, reader := bcUser(t, pg, "Admin"), bcUser(t, pg, "R") + + past := time.Now().Add(-time.Hour) + b, err := svc.Create(ctx, admin, CreateParams{ + Title: "Old news", Body: "Yesterday's announcement.", ExpiresAt: &past, + }) + if err != nil { + t.Fatal(err) + } + cleanupBroadcast(t, pg, b.ID) + if _, err := svc.Publish(ctx, b.ID); err != nil { + t.Fatal(err) + } + + if got, _ := svc.Pending(ctx, reader); got != nil { + t.Errorf("an expired announcement was offered: %+v", got) + } +} + +// Only the newest announcement is offered: two modals stacked on launch is a +// worse experience than missing the older one. +func TestBroadcast_OffersNewestFirst(t *testing.T) { + svc, pg := testBroadcastService(t) + ctx := context.Background() + admin, reader := bcUser(t, pg, "Admin"), bcUser(t, pg, "R") + + older, err := svc.Create(ctx, admin, CreateParams{Title: "First", Body: "one"}) + if err != nil { + t.Fatal(err) + } + cleanupBroadcast(t, pg, older.ID) + if _, err := svc.Publish(ctx, older.ID); err != nil { + t.Fatal(err) + } + + newer, err := svc.Create(ctx, admin, CreateParams{Title: "Second", Body: "two"}) + if err != nil { + t.Fatal(err) + } + cleanupBroadcast(t, pg, newer.ID) + if _, err := svc.Publish(ctx, newer.ID); err != nil { + t.Fatal(err) + } + + got, err := svc.Pending(ctx, reader) + if err != nil { + t.Fatal(err) + } + if got == nil || got.ID != newer.ID { + t.Errorf("expected the newest announcement, got %+v", got) + } +} diff --git a/migrations/0052_broadcasts.down.sql b/migrations/0052_broadcasts.down.sql new file mode 100644 index 0000000..1679645 --- /dev/null +++ b/migrations/0052_broadcasts.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS broadcast_views; +DROP TABLE IF EXISTS broadcasts; diff --git a/migrations/0052_broadcasts.up.sql b/migrations/0052_broadcasts.up.sql new file mode 100644 index 0000000..5c08ff8 --- /dev/null +++ b/migrations/0052_broadcasts.up.sql @@ -0,0 +1,44 @@ +-- Admin broadcasts — product updates and announcements, shown once per user. +-- +-- On "no history": these deliberately create no notification rows and never +-- appear in the tray. But "exactly once per user" cannot be done without +-- storing something — the app has to know whether YOU have seen it. So two +-- tables: the announcement, and who has dismissed it. Nothing per-user is +-- written until that person actually sees it. +CREATE TABLE IF NOT EXISTS broadcasts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + title VARCHAR(200) NOT NULL, + body TEXT NOT NULL DEFAULT '', + -- Formatting as offset/length spans, exactly like chat messages: the same + -- parser produces them and the same component renders them, so no HTML is + -- stored and nothing needs sanitising at render time. + entities JSONB NOT NULL DEFAULT '[]'::jsonb, + -- R2 key for the banner. Public bucket: an announcement is not private, + -- and a signed URL would expire while the modal sat open. + banner_key TEXT, + cta_label VARCHAR(60), + -- App-relative path or absolute http(s) URL; validated in the service. + cta_url TEXT, + created_by UUID REFERENCES users(id) ON DELETE SET NULL, + -- NULL = draft. Nothing is delivered until this is set. + published_at TIMESTAMPTZ, + -- After this, the announcement stops being offered. This is what keeps a + -- months-old update from greeting someone who signs up tomorrow. + expires_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- "What is live right now?" — the app's pending lookup and the admin list. +CREATE INDEX IF NOT EXISTS broadcasts_live_idx + ON broadcasts (published_at DESC) WHERE published_at IS NOT NULL; + +-- One row per person who has seen an announcement. Written on dismissal, which +-- is what makes "at most once" true across devices — localStorage would show +-- it again on every new browser. +CREATE TABLE IF NOT EXISTS broadcast_views ( + broadcast_id UUID NOT NULL REFERENCES broadcasts(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + seen_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (broadcast_id, user_id) +);