fixes

internal/app/broadcast/service.go

@@ -1,10 +1,35 @@
 // 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.
+// 777000) to every user or a hand-picked list.
+//
+// A "selected" campaign's recipient rows are all inserted at creation, since
+// that list is bounded by domain.MaxBroadcastSelectedRecipients. An "all"
+// campaign instead only snapshots the current max eligible user id at
+// creation, and store.BroadcastStore.MaterializeBroadcastRecipients walks
+// that range incrementally, a bounded batch per worker cycle -- so creating
+// a campaign for a large user base is a single cheap insert, not one giant
+// blocking transaction.
+//
+// Delivery is a lease-based claim cycle (store.BroadcastStore.
+// ClaimBroadcastRecipients/CompleteBroadcastRecipient/ReleaseBroadcastRecipient):
+// a worker leases a bounded batch of eligible rows for a fixed duration,
+// delivers each one, and closes it out. A lease that is never renewed simply
+// expires, so a worker crash mid-cycle cannot strand rows in 'processing'
+// forever, and a future multi-instance worker can run the same cycle
+// concurrently without two instances ever believing they hold the same
+// row's lease at once.
+//
+// Delivery itself goes through messageSender.SendPrivateText -- the same
+// store-layer send path internal/app/bots's sendServiceBotReplyResult calls
+// directly (bypassing the auth-checked app.messages.Service wrapper, which
+// requires SenderUserID == the authenticated caller) -- rather than
+// duplicating message/pts/dispatch-outbox creation here. SendPrivateText's
+// random_id dedup is what actually closes the small race a lease alone
+// leaves open: if a lease expires and gets reclaimed while the original
+// holder's send is still in flight, both attempts use the same
+// (broadcastID, userID)-derived random id (see stableBroadcastRandomID), so
+// the store resolves them to the very same message instead of sending
+// twice, no matter which claim ends up recording it.
 package broadcast
 
 import (
@@ -13,6 +38,8 @@ import (
 	"hash/fnv"
 	"strconv"
 	"strings"
+	"time"
+	"unicode/utf8"
 
 	"go.uber.org/zap"
 
@@ -21,10 +48,7 @@ import (
 )
 
 // 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).
+// store.MessageStore: sending a message with an arbitrary SenderUserID.
 type messageSender interface {
 	SendPrivateText(ctx context.Context, req domain.SendPrivateTextRequest) (domain.SendPrivateTextResult, error)
 }
@@ -69,22 +93,80 @@ func WithLogger(log *zap.Logger) Option {
 // 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) {
+func normalizeRequest(message string, mode domain.BroadcastTargetMode, selectedUserIDs []int64) (string, []int64, error) {
+	message = strings.TrimSpace(message)
+	if message == "" {
+		return "", nil, domain.ErrBroadcastMessageEmpty
+	}
+	if !utf8.ValidString(message) || len(message) > domain.MaxBroadcastMessageBytes {
+		return "", nil, domain.ErrBroadcastMessageTooLong
+	}
+	switch mode {
+	case domain.BroadcastTargetAll:
+		if len(selectedUserIDs) != 0 {
+			return "", nil, domain.ErrBroadcastInvalid
+		}
+		return message, nil, nil
+	case domain.BroadcastTargetSelected:
+		if len(selectedUserIDs) == 0 {
+			return "", nil, domain.ErrBroadcastNoRecipients
+		}
+		if len(selectedUserIDs) > domain.MaxBroadcastSelectedRecipients {
+			return "", nil, domain.ErrBroadcastInvalid
+		}
+		seen := make(map[int64]struct{}, len(selectedUserIDs))
+		ids := make([]int64, 0, len(selectedUserIDs))
+		for _, userID := range selectedUserIDs {
+			if userID <= 0 || domain.IsSystemUserID(userID) {
+				return "", nil, domain.ErrBroadcastRecipientInvalid
+			}
+			if _, ok := seen[userID]; ok {
+				return "", nil, domain.ErrBroadcastRecipientInvalid
+			}
+			seen[userID] = struct{}{}
+			ids = append(ids, userID)
+		}
+		return message, ids, nil
+	default:
+		return "", nil, domain.ErrBroadcastInvalid
+	}
+}
+
+// Preview validates and counts a campaign's intended recipient set without
+// creating anything, so an admin UI can show "this will reach N users"
+// before committing.
+func (s *Service) Preview(ctx context.Context, message string, mode domain.BroadcastTargetMode, selectedUserIDs []int64) (int64, error) {
+	if s == nil || s.store == nil {
+		return 0, fmt.Errorf("broadcast store is not configured")
+	}
+	_, ids, err := normalizeRequest(message, mode, selectedUserIDs)
+	if err != nil {
+		return 0, err
+	}
+	return s.store.PreviewBroadcastRecipients(ctx, mode, ids)
+}
+
+// Create validates and snapshots a new broadcast, then returns immediately:
+// delivery (and, for "all" mode, recipient enumeration itself) happens
+// asynchronously via the Worker's RunCycle, so this never blocks an admin
+// HTTP request on however many recipients there are.
+//
+// Entities are derived automatically from the plain-text message (mentions,
+// hashtags, cashtags, bot commands -- see domain.DetectAutomaticMessageEntities),
+// not operator-composed: there is currently no admin UI for hand-authoring
+// bold/italic/link spans on a broadcast, so this only gets a broadcast the
+// same clickable-entity rendering any other plain-text message with an
+// @mention or #hashtag already gets.
+func (s *Service) Create(ctx context.Context, message string, mode domain.BroadcastTargetMode, selectedUserIDs []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
+	message, ids, err := normalizeRequest(message, mode, selectedUserIDs)
+	if err != nil {
+		return domain.Broadcast{}, err
 	}
-	if targetMode != domain.BroadcastTargetAll && targetMode != domain.BroadcastTargetSelected {
-		return domain.Broadcast{}, domain.ErrBroadcastInvalid
-	}
-	return s.store.CreateBroadcast(ctx, message, targetMode, recipientUserIDs, createdBy)
+	entities := domain.DetectAutomaticMessageEntities(message, nil)
+	return s.store.CreateBroadcast(ctx, message, entities, mode, ids, strings.TrimSpace(createdBy))
 }
 
 // List pages broadcasts newest-first.
@@ -103,51 +185,80 @@ func (s *Service) Get(ctx context.Context, id int64) (domain.Broadcast, bool, er
 	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
+// CycleResult reports one worker cycle's outcome.
+type CycleResult struct {
+	Materialized int
+	Claimed      int
+	Sent         int
+	Failed       int
+}
+
+// RunCycle advances "all"-mode enumeration by up to materializeBatch rows,
+// then claims up to deliveryBatch eligible recipient rows under leaseToken
+// for lease, delivering each one via SendPrivateText. One recipient's
+// failure (blocked account, deleted account, transient error) never blocks
+// the rest of the batch.
+func (s *Service) RunCycle(ctx context.Context, leaseToken string, materializeBatch, deliveryBatch int, lease time.Duration) (CycleResult, error) {
+	var result CycleResult
+	if !s.Ready() {
+		return result, nil
 	}
-	pending, err := s.store.PendingBroadcastRecipients(ctx, limit)
+	materialized, err := s.store.MaterializeBroadcastRecipients(ctx, materializeBatch)
 	if err != nil {
-		return 0, err
+		return result, err
 	}
-	for _, recipient := range pending {
+	result.Materialized = materialized
+	claims, err := s.store.ClaimBroadcastRecipients(ctx, leaseToken, deliveryBatch, lease)
+	if err != nil {
+		return result, err
+	}
+	result.Claimed = len(claims)
+	for _, claim := range claims {
 		if err := ctx.Err(); err != nil {
-			return sent, err
+			return result, 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))
+		if err := s.deliverClaim(ctx, claim); err != nil {
+			result.Failed++
+			if releaseErr := s.store.ReleaseBroadcastRecipient(ctx, claim, err.Error()); releaseErr != nil {
+				s.log.Warn("release broadcast recipient failed",
+					zap.Int64("recipient_id", claim.RecipientID),
+					zap.Int64("broadcast_id", claim.BroadcastID),
+					zap.Error(releaseErr))
 			}
 			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++
+		result.Sent++
 	}
-	return sent, nil
+	return result, nil
+}
+
+func (s *Service) deliverClaim(ctx context.Context, claim store.BroadcastRecipientClaim) error {
+	send, err := s.messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{
+		SenderUserID:    domain.OfficialSystemUserID,
+		RecipientUserID: claim.UserID,
+		// A stable id derived from (broadcast, recipient) makes redelivering
+		// this exact row idempotent at the store layer's random_id dedup: if
+		// this claim's lease expires and gets reclaimed while a prior send is
+		// still in flight, both resolve to the same message instead of
+		// sending twice.
+		RandomID: stableBroadcastRandomID(claim.BroadcastID, claim.UserID),
+		Message:  claim.Message,
+		Entities: claim.Entities,
+	})
+	if err != nil {
+		return err
+	}
+	msg := send.RecipientMessage
+	if err := s.store.CompleteBroadcastRecipient(ctx, claim, msg.UID, msg.ID, msg.Pts); err != nil {
+		return err
+	}
+	return 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.
+// re-processing the same recipient row (after a lease is reclaimed, before
+// it was recorded 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)))

internal/app/broadcast/service_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"telesrv/internal/domain"
 	"telesrv/internal/store/memory"
@@ -12,6 +13,7 @@ import (
 type fakeSender struct {
 	sent    []domain.SendPrivateTextRequest
 	failFor map[int64]bool // fail every send to this recipient user id
+	nextID  int
 }
 
 func (f *fakeSender) SendPrivateText(_ context.Context, req domain.SendPrivateTextRequest) (domain.SendPrivateTextResult, error) {
@@ -19,14 +21,21 @@ func (f *fakeSender) SendPrivateText(_ context.Context, req domain.SendPrivateTe
 		return domain.SendPrivateTextResult{}, errors.New("simulated send failure")
 	}
 	f.sent = append(f.sent, req)
-	return domain.SendPrivateTextResult{}, nil
+	f.nextID++
+	return domain.SendPrivateTextResult{
+		RecipientMessage: domain.Message{
+			ID:  f.nextID,
+			UID: int64(f.nextID),
+			Pts: f.nextID,
+		},
+	}, 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) {
+	if _, err := svc.Create(ctx, "  ", domain.BroadcastTargetAll, nil, "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) {
@@ -35,21 +44,33 @@ func TestCreateValidatesInput(t *testing.T) {
 	if _, err := svc.Create(ctx, "hi", domain.BroadcastTargetSelected, nil, "admin"); !errors.Is(err, domain.ErrBroadcastNoRecipients) {
 		t.Fatalf("no recipients: err = %v, want ErrBroadcastNoRecipients", err)
 	}
+	tooMany := make([]int64, domain.MaxBroadcastSelectedRecipients+1)
+	for i := range tooMany {
+		tooMany[i] = int64(1000 + i)
+	}
+	if _, err := svc.Create(ctx, "hi", domain.BroadcastTargetSelected, tooMany, "admin"); !errors.Is(err, domain.ErrBroadcastInvalid) {
+		t.Fatalf("too many recipients: err = %v, want ErrBroadcastInvalid", err)
+	}
 
-	created, err := svc.Create(ctx, "  News! ", domain.BroadcastTargetSelected, []int64{10, 20, 20}, "admin")
+	// A duplicate id in the selected list is rejected outright, not silently
+	// collapsed: the caller's list should already be a set.
+	if _, err := svc.Create(ctx, "hi", domain.BroadcastTargetSelected, []int64{10, 20, 20}, "admin"); !errors.Is(err, domain.ErrBroadcastRecipientInvalid) {
+		t.Fatalf("duplicate recipient: err = %v, want ErrBroadcastRecipientInvalid", err)
+	}
+
+	created, err := svc.Create(ctx, "  News! ", domain.BroadcastTargetSelected, []int64{10, 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)
+	if created.TargetCount != 2 {
+		t.Fatalf("TargetCount = %d, want 2", created.TargetCount)
 	}
 }
 
-func TestRunSendCycleDeliversAndCounts(t *testing.T) {
+func TestRunCycleDeliversAndCounts(t *testing.T) {
 	store := memory.NewBroadcastStore()
 	sender := &fakeSender{}
 	svc := NewService(store, WithMessageSender(sender))
@@ -60,12 +81,12 @@ func TestRunSendCycleDeliversAndCounts(t *testing.T) {
 		t.Fatalf("Create: %v", err)
 	}
 
-	sent, err := svc.RunSendCycle(ctx, 10)
+	result, err := svc.RunCycle(ctx, "lease-1", 100, 10, 30*time.Second)
 	if err != nil {
-		t.Fatalf("RunSendCycle: %v", err)
+		t.Fatalf("RunCycle: %v", err)
 	}
-	if sent != 3 {
-		t.Fatalf("sent = %d, want 3", sent)
+	if result.Sent != 3 {
+		t.Fatalf("Sent = %d, want 3", result.Sent)
 	}
 	if len(sender.sent) != 3 {
 		t.Fatalf("sender received %d sends, want 3", len(sender.sent))
@@ -88,16 +109,16 @@ func TestRunSendCycleDeliversAndCounts(t *testing.T) {
 	}
 
 	// A second cycle finds nothing left pending.
-	sent, err = svc.RunSendCycle(ctx, 10)
+	result, err = svc.RunCycle(ctx, "lease-2", 100, 10, 30*time.Second)
 	if err != nil {
-		t.Fatalf("RunSendCycle (second): %v", err)
+		t.Fatalf("RunCycle (second): %v", err)
 	}
-	if sent != 0 {
-		t.Fatalf("second cycle sent = %d, want 0 (nothing pending)", sent)
+	if result.Claimed != 0 {
+		t.Fatalf("second cycle claimed = %d, want 0 (nothing pending)", result.Claimed)
 	}
 }
 
-func TestRunSendCycleRetriesThenTerminatesFailures(t *testing.T) {
+func TestRunCycleRetriesThenTerminatesFailures(t *testing.T) {
 	store := memory.NewBroadcastStore()
 	sender := &fakeSender{failFor: map[int64]bool{999: true}}
 	svc := NewService(store, WithMessageSender(sender))
@@ -110,22 +131,49 @@ func TestRunSendCycleRetriesThenTerminatesFailures(t *testing.T) {
 	// 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)
+		result, err := svc.RunCycle(ctx, "lease", 100, 10, 30*time.Second)
 		if err != nil {
-			t.Fatalf("RunSendCycle attempt %d: %v", i+1, err)
+			t.Fatalf("RunCycle attempt %d: %v", i+1, err)
 		}
-		if sent != 0 {
-			t.Fatalf("attempt %d: sent = %d, want 0 (always fails)", i+1, sent)
+		if result.Sent != 0 || result.Failed != 1 {
+			t.Fatalf("attempt %d: sent=%d failed=%d, want sent=0 failed=1 (always fails)", i+1, result.Sent, result.Failed)
 		}
 	}
 
-	// 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)
+	// One more cycle: the row is now terminal ('failed'), so nothing is left
+	// to claim.
+	result, err := svc.RunCycle(ctx, "lease-final", 100, 10, 30*time.Second)
 	if err != nil {
-		t.Fatalf("PendingBroadcastRecipients: %v", err)
+		t.Fatalf("RunCycle (final): %v", err)
 	}
-	if len(pending) != 0 {
-		t.Fatalf("pending = %+v, want empty (recipient should be terminally failed)", pending)
+	if result.Claimed != 0 {
+		t.Fatalf("final cycle claimed = %d, want 0 (recipient should be terminally failed)", result.Claimed)
+	}
+}
+
+func TestCreateAllModeSnapshotsWithoutExplicitIDs(t *testing.T) {
+	store := memory.NewBroadcastStore()
+	store.SeedEligibleUsers([]int64{1, 2, 3})
+	sender := &fakeSender{}
+	svc := NewService(store, WithMessageSender(sender))
+	ctx := context.Background()
+
+	created, err := svc.Create(ctx, "hello all", domain.BroadcastTargetAll, nil, "admin")
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	if created.TargetMode != domain.BroadcastTargetAll {
+		t.Fatalf("TargetMode = %q, want all", created.TargetMode)
+	}
+
+	result, err := svc.RunCycle(ctx, "lease", 100, 100, 30*time.Second)
+	if err != nil {
+		t.Fatalf("RunCycle: %v", err)
+	}
+	if result.Materialized != 3 {
+		t.Fatalf("Materialized = %d, want 3", result.Materialized)
+	}
+	if result.Sent != 3 {
+		t.Fatalf("Sent = %d, want 3", result.Sent)
 	}
 }

internal/app/broadcast/worker.go

@@ -2,59 +2,69 @@ package broadcast
 
 import (
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"time"
 
 	"go.uber.org/zap"
 )
 
-// defaultInterval/defaultBatch match the shipped
-// TELESRV_BROADCAST_WORKER_INTERVAL/_BATCH defaults.
-const (
-	defaultInterval = 3 * time.Second
-	defaultBatch    = 50
-)
+// WorkerConfig tunes the periodic materialize+delivery cycle. Non-positive
+// or out-of-range fields fall back to the defaults below (matching the
+// shipped TELESRV_BROADCAST_WORKER_* defaults).
+type WorkerConfig struct {
+	Interval         time.Duration
+	Lease            time.Duration
+	MaterializeBatch int
+	DeliveryBatch    int
+}
 
-// Worker drains the broadcast delivery outbox.
+// Worker drains the broadcast delivery outbox and, for "all"-mode
+// campaigns, the recipient-enumeration backlog.
 //
-// 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.
+// A broadcast is created together with only its target snapshot, never with
+// the enumeration or the sends themselves: an admin creating a broadcast for
+// every user must not wait on however long that would take. Both
+// materialization and delivery are 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
+	service *Service
+	config  WorkerConfig
+	log     *zap.Logger
 }
 
-// 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()
+// NewWorker creates the periodic worker.
+func NewWorker(service *Service, config WorkerConfig, log *zap.Logger) *Worker {
+	if config.Interval <= 0 {
+		config.Interval = 3 * time.Second
 	}
-	if interval <= 0 {
-		interval = defaultInterval
+	if config.Lease <= 0 {
+		config.Lease = 30 * time.Second
 	}
-	if batch <= 0 {
-		batch = defaultBatch
+	if config.MaterializeBatch <= 0 || config.MaterializeBatch > 1000 {
+		config.MaterializeBatch = 200
 	}
-	return &Worker{service: service, logger: logger, interval: interval, batch: batch}
+	if config.DeliveryBatch <= 0 || config.DeliveryBatch > 500 {
+		config.DeliveryBatch = 50
+	}
+	if log == nil {
+		log = zap.NewNop()
+	}
+	return &Worker{service: service, config: config, log: log}
 }
 
-// Run delivers one batch immediately and then on every tick until ctx is
+// Run advances one cycle 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")
+	if w == nil || w.service == nil || !w.service.Ready() {
+		if w != nil && w.log != nil {
+			w.log.Info("broadcast delivery worker disabled: not configured")
+		}
 		return
 	}
 	w.runOnce(ctx)
-	ticker := time.NewTicker(w.interval)
+	ticker := time.NewTicker(w.config.Interval)
 	defer ticker.Stop()
 	for {
 		select {
@@ -67,18 +77,23 @@ func (w *Worker) Run(ctx context.Context) {
 }
 
 func (w *Worker) runOnce(ctx context.Context) {
-	if w == nil || w.service == nil {
+	var tokenBytes [16]byte
+	if _, err := rand.Read(tokenBytes[:]); err != nil {
+		w.log.Error("generate broadcast lease token", zap.Error(err))
 		return
 	}
-	sent, err := w.service.RunSendCycle(ctx, w.batch)
+	result, err := w.service.RunCycle(ctx, hex.EncodeToString(tokenBytes[:]), w.config.MaterializeBatch, w.config.DeliveryBatch, w.config.Lease)
 	if err != nil {
-		if ctx.Err() != nil {
-			return
+		if ctx.Err() == nil {
+			w.log.Warn("broadcast delivery cycle failed", zap.Error(err))
 		}
-		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))
+	if result.Materialized > 0 || result.Claimed > 0 {
+		w.log.Info("broadcast delivery cycle completed",
+			zap.Int("materialized", result.Materialized),
+			zap.Int("claimed", result.Claimed),
+			zap.Int("sent", result.Sent),
+			zap.Int("failed", result.Failed))
 	}
 }