added ability to broadcast

internal/app/broadcast/service.go

@@ -0,0 +1,159 @@
+// Package broadcast implements admin-triggered system message campaigns:
+// sending a message from the official system account (domain.OfficialSystemUserID,
+// 777000) to every user or a hand-picked list. Delivery is a durable outbox
+// (store.BroadcastStore's recipient rows) drained by a periodic Worker,
+// mirroring internal/app/verification's notification outbox -- the admin
+// action only snapshots the recipient list and returns, never sending
+// potentially thousands of messages inline within one HTTP request.
+package broadcast
+
+import (
+	"context"
+	"fmt"
+	"hash/fnv"
+	"strconv"
+	"strings"
+
+	"go.uber.org/zap"
+
+	"telesrv/internal/domain"
+	"telesrv/internal/store"
+)
+
+// messageSender is the narrow port this package needs from
+// store.MessageStore: sending a message with an arbitrary SenderUserID, the
+// way internal/app/bots's sendServiceBotReplyResult calls it directly at the
+// store layer rather than through the auth-checked app.messages.Service
+// wrapper (which requires SenderUserID == the authenticated caller).
+type messageSender interface {
+	SendPrivateText(ctx context.Context, req domain.SendPrivateTextRequest) (domain.SendPrivateTextResult, error)
+}
+
+// Service creates broadcasts and drains their delivery outbox.
+type Service struct {
+	store    store.BroadcastStore
+	messages messageSender
+	log      *zap.Logger
+}
+
+// Option adjusts an optional Service dependency.
+type Option func(*Service)
+
+// NewService builds the broadcast service.
+func NewService(st store.BroadcastStore, opts ...Option) *Service {
+	s := &Service{store: st, log: zap.NewNop()}
+	for _, opt := range opts {
+		opt(s)
+	}
+	return s
+}
+
+// WithMessageSender injects the store used to actually deliver a message.
+func WithMessageSender(m messageSender) Option {
+	return func(s *Service) {
+		if m != nil {
+			s.messages = m
+		}
+	}
+}
+
+// WithLogger injects a logger (default zap.NewNop()).
+func WithLogger(log *zap.Logger) Option {
+	return func(s *Service) {
+		if log != nil {
+			s.log = log
+		}
+	}
+}
+
+// Ready reports whether both the store and the sender are wired.
+func (s *Service) Ready() bool { return s != nil && s.store != nil && s.messages != nil }
+
+// Create validates and snapshots a new broadcast's recipient set, then
+// returns immediately: delivery happens asynchronously via RunSendCycle, so
+// this never blocks an admin HTTP request on however many recipients there
+// are.
+func (s *Service) Create(ctx context.Context, message string, targetMode domain.BroadcastTargetMode, recipientUserIDs []int64, createdBy string) (domain.Broadcast, error) {
+	if s == nil || s.store == nil {
+		return domain.Broadcast{}, fmt.Errorf("broadcast store is not configured")
+	}
+	message = strings.TrimSpace(message)
+	if message == "" {
+		return domain.Broadcast{}, domain.ErrBroadcastMessageEmpty
+	}
+	if targetMode != domain.BroadcastTargetAll && targetMode != domain.BroadcastTargetSelected {
+		return domain.Broadcast{}, domain.ErrBroadcastInvalid
+	}
+	return s.store.CreateBroadcast(ctx, message, targetMode, recipientUserIDs, createdBy)
+}
+
+// List pages broadcasts newest-first.
+func (s *Service) List(ctx context.Context, beforeID int64, limit int) ([]domain.Broadcast, bool, error) {
+	if s == nil || s.store == nil {
+		return nil, false, nil
+	}
+	return s.store.ListBroadcasts(ctx, beforeID, limit)
+}
+
+// Get returns one broadcast.
+func (s *Service) Get(ctx context.Context, id int64) (domain.Broadcast, bool, error) {
+	if s == nil || s.store == nil {
+		return domain.Broadcast{}, false, nil
+	}
+	return s.store.BroadcastByID(ctx, id)
+}
+
+// RunSendCycle drains up to limit pending recipient rows, sending each from
+// domain.OfficialSystemUserID. One recipient's failure (blocked account,
+// deleted account, transient error) never blocks the rest of the batch.
+func (s *Service) RunSendCycle(ctx context.Context, limit int) (sent int, err error) {
+	if s == nil || !s.Ready() {
+		return 0, nil
+	}
+	pending, err := s.store.PendingBroadcastRecipients(ctx, limit)
+	if err != nil {
+		return 0, err
+	}
+	for _, recipient := range pending {
+		if err := ctx.Err(); err != nil {
+			return sent, err
+		}
+		_, sendErr := s.messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{
+			SenderUserID:    domain.OfficialSystemUserID,
+			RecipientUserID: recipient.UserID,
+			// A stable id derived from (broadcast, recipient) makes reprocessing
+			// this exact row idempotent at the store layer's random_id dedup,
+			// instead of risking a duplicate message if this worker crashes
+			// between sending and marking the row delivered.
+			RandomID: stableBroadcastRandomID(recipient.BroadcastID, recipient.UserID),
+			Message:  recipient.Message,
+		})
+		if sendErr != nil {
+			if markErr := s.store.MarkBroadcastRecipientFailed(ctx, recipient.RecipientID, sendErr.Error()); markErr != nil {
+				s.log.Warn("mark broadcast recipient failed",
+					zap.Int64("recipient_id", recipient.RecipientID), zap.Error(markErr))
+			}
+			continue
+		}
+		if markErr := s.store.MarkBroadcastRecipientSent(ctx, recipient.RecipientID); markErr != nil {
+			s.log.Warn("mark broadcast recipient sent",
+				zap.Int64("recipient_id", recipient.RecipientID), zap.Error(markErr))
+			continue
+		}
+		sent++
+	}
+	return sent, nil
+}
+
+// stableBroadcastRandomID derives a random_id from (broadcastID, userID) so
+// re-processing the same recipient row (after a crash, before it was marked
+// delivered) resolves to the same send instead of a duplicate message.
+func stableBroadcastRandomID(broadcastID, userID int64) int64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(strconv.FormatInt(broadcastID, 10) + ":" + strconv.FormatInt(userID, 10)))
+	v := int64(h.Sum64())
+	if v == 0 {
+		v = 1
+	}
+	return v
+}

internal/app/broadcast/service_test.go

@@ -0,0 +1,131 @@
+package broadcast
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"telesrv/internal/domain"
+	"telesrv/internal/store/memory"
+)
+
+type fakeSender struct {
+	sent    []domain.SendPrivateTextRequest
+	failFor map[int64]bool // fail every send to this recipient user id
+}
+
+func (f *fakeSender) SendPrivateText(_ context.Context, req domain.SendPrivateTextRequest) (domain.SendPrivateTextResult, error) {
+	if f.failFor[req.RecipientUserID] {
+		return domain.SendPrivateTextResult{}, errors.New("simulated send failure")
+	}
+	f.sent = append(f.sent, req)
+	return domain.SendPrivateTextResult{}, nil
+}
+
+func TestCreateValidatesInput(t *testing.T) {
+	svc := NewService(memory.NewBroadcastStore(), WithMessageSender(&fakeSender{}))
+	ctx := context.Background()
+
+	if _, err := svc.Create(ctx, "  ", domain.BroadcastTargetAll, []int64{1}, "admin"); !errors.Is(err, domain.ErrBroadcastMessageEmpty) {
+		t.Fatalf("empty message: err = %v, want ErrBroadcastMessageEmpty", err)
+	}
+	if _, err := svc.Create(ctx, "hi", domain.BroadcastTargetMode("bogus"), []int64{1}, "admin"); !errors.Is(err, domain.ErrBroadcastInvalid) {
+		t.Fatalf("bad target mode: err = %v, want ErrBroadcastInvalid", err)
+	}
+	if _, err := svc.Create(ctx, "hi", domain.BroadcastTargetSelected, nil, "admin"); !errors.Is(err, domain.ErrBroadcastNoRecipients) {
+		t.Fatalf("no recipients: err = %v, want ErrBroadcastNoRecipients", err)
+	}
+
+	created, err := svc.Create(ctx, "  News! ", domain.BroadcastTargetSelected, []int64{10, 20, 20}, "admin")
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	if created.Message != "News!" {
+		t.Fatalf("Message = %q, want trimmed %q", created.Message, "News!")
+	}
+	// The duplicate recipient (20 twice) collapses to one row.
+	if created.TotalCount != 2 {
+		t.Fatalf("TotalCount = %d, want 2 (duplicate recipient collapsed)", created.TotalCount)
+	}
+}
+
+func TestRunSendCycleDeliversAndCounts(t *testing.T) {
+	store := memory.NewBroadcastStore()
+	sender := &fakeSender{}
+	svc := NewService(store, WithMessageSender(sender))
+	ctx := context.Background()
+
+	created, err := svc.Create(ctx, "Update available", domain.BroadcastTargetSelected, []int64{101, 102, 103}, "admin")
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	sent, err := svc.RunSendCycle(ctx, 10)
+	if err != nil {
+		t.Fatalf("RunSendCycle: %v", err)
+	}
+	if sent != 3 {
+		t.Fatalf("sent = %d, want 3", sent)
+	}
+	if len(sender.sent) != 3 {
+		t.Fatalf("sender received %d sends, want 3", len(sender.sent))
+	}
+	for _, req := range sender.sent {
+		if req.SenderUserID != domain.OfficialSystemUserID {
+			t.Fatalf("SenderUserID = %d, want OfficialSystemUserID (%d)", req.SenderUserID, domain.OfficialSystemUserID)
+		}
+		if req.Message != "Update available" {
+			t.Fatalf("Message = %q, want %q", req.Message, "Update available")
+		}
+	}
+
+	got, found, err := svc.Get(ctx, created.ID)
+	if err != nil || !found {
+		t.Fatalf("Get: found=%v err=%v", found, err)
+	}
+	if got.SentCount != 3 || got.FailedCount != 0 {
+		t.Fatalf("counts = sent:%d failed:%d, want sent:3 failed:0", got.SentCount, got.FailedCount)
+	}
+
+	// A second cycle finds nothing left pending.
+	sent, err = svc.RunSendCycle(ctx, 10)
+	if err != nil {
+		t.Fatalf("RunSendCycle (second): %v", err)
+	}
+	if sent != 0 {
+		t.Fatalf("second cycle sent = %d, want 0 (nothing pending)", sent)
+	}
+}
+
+func TestRunSendCycleRetriesThenTerminatesFailures(t *testing.T) {
+	store := memory.NewBroadcastStore()
+	sender := &fakeSender{failFor: map[int64]bool{999: true}}
+	svc := NewService(store, WithMessageSender(sender))
+	ctx := context.Background()
+
+	if _, err := svc.Create(ctx, "will fail", domain.BroadcastTargetSelected, []int64{999}, "admin"); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	// Run one cycle per attempt, up to the cap; the row must stay pending
+	// (retried) below the cap and become terminal at it.
+	for i := 0; i < domain.MaxBroadcastRecipientAttempts; i++ {
+		sent, err := svc.RunSendCycle(ctx, 10)
+		if err != nil {
+			t.Fatalf("RunSendCycle attempt %d: %v", i+1, err)
+		}
+		if sent != 0 {
+			t.Fatalf("attempt %d: sent = %d, want 0 (always fails)", i+1, sent)
+		}
+	}
+
+	// One more cycle: the row is now terminal ('failed'), so PendingBroadcastRecipients
+	// must not return it, and RunSendCycle finds nothing left to attempt.
+	pending, err := store.PendingBroadcastRecipients(ctx, 10)
+	if err != nil {
+		t.Fatalf("PendingBroadcastRecipients: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("pending = %+v, want empty (recipient should be terminally failed)", pending)
+	}
+}

internal/app/broadcast/worker.go

@@ -0,0 +1,84 @@
+package broadcast
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// defaultInterval/defaultBatch match the shipped
+// TELESRV_BROADCAST_WORKER_INTERVAL/_BATCH defaults.
+const (
+	defaultInterval = 3 * time.Second
+	defaultBatch    = 50
+)
+
+// Worker drains the broadcast delivery outbox.
+//
+// A broadcast is created together with its recipient snapshot, never with the
+// sends themselves: an admin creating a broadcast for every user must not
+// wait on however long that takes. Delivery is therefore a separate,
+// retrying cycle over durable rows, and this worker is only its cadence.
+type Worker struct {
+	service  *Service
+	logger   *zap.Logger
+	interval time.Duration
+	batch    int
+}
+
+// NewWorker creates the periodic delivery worker. Non-positive
+// interval/batch fall back to the shipped defaults.
+func NewWorker(service *Service, logger *zap.Logger, interval time.Duration, batch int) *Worker {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	if interval <= 0 {
+		interval = defaultInterval
+	}
+	if batch <= 0 {
+		batch = defaultBatch
+	}
+	return &Worker{service: service, logger: logger, interval: interval, batch: batch}
+}
+
+// Run delivers one batch immediately and then on every tick until ctx is
+// done. A not-ready service (missing store/sender) exits immediately with
+// one explicit log line instead of ticking forever over a no-op.
+func (w *Worker) Run(ctx context.Context) {
+	if w == nil {
+		return
+	}
+	if !w.service.Ready() {
+		w.logger.Info("broadcast delivery worker disabled: not configured")
+		return
+	}
+	w.runOnce(ctx)
+	ticker := time.NewTicker(w.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			w.runOnce(ctx)
+		}
+	}
+}
+
+func (w *Worker) runOnce(ctx context.Context) {
+	if w == nil || w.service == nil {
+		return
+	}
+	sent, err := w.service.RunSendCycle(ctx, w.batch)
+	if err != nil {
+		if ctx.Err() != nil {
+			return
+		}
+		w.logger.Warn("broadcast delivery cycle failed", zap.Int("sent", sent), zap.Int("batch", w.batch), zap.Error(err))
+		return
+	}
+	if sent > 0 {
+		w.logger.Info("broadcast delivery cycle completed", zap.Int("sent", sent), zap.Int("batch", w.batch))
+	}
+}