feat: broadcast

2026-08-29 18:34:07 +05:30
parent 44120efdec
commit f65f552c87
9 changed files with 869 additions and 0 deletions
@@ -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
+}
@@ -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)
+	}
+}