diplomail (Stage C): paid-tier broadcast + multi-game + cleanup

Closes out the producer-side of the diplomail surface. Paid-tier
players can fan out one personal message to the rest of the active
roster (gated on entitlement_snapshots.is_paid). Site admins gain a
multi-game broadcast (POST /admin/mail/broadcast with `selected` /
`all_running` scopes) and the bulk-purge endpoint that wipes
diplomail rows tied to games finished more than N years ago. An
admin listing (GET /admin/mail/messages) rounds out the
observability surface.

EntitlementReader and GameLookup are new narrow deps wired from
`*user.Service` and `*lobby.Service` in cmd/backend/main; the lobby
service grows a one-off `ListFinishedGamesBefore` helper for the
cleanup path (the cache evicts terminal-state games so the cache
walk is not enough). Stage D will swap LangUndetermined for an
actual body-language detector and add the translation cache.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/internal/diplomail/README.md

@@ -16,7 +16,7 @@ purge, and the language-detection / translation cache.
 |-------|-------|--------|
 | A | Schema, personal single-recipient send / read / delete, unread badge, push event with body-language `und` | shipped |
 | B | Owner / admin sends + lifecycle hooks (paused, cancelled, kick); strict soft-access for kicked players | shipped |
-| C | Paid-tier personal broadcast + admin multi-game broadcast + bulk purge | planned |
+| C | Paid-tier personal broadcast + admin multi-game broadcast + bulk purge + admin observability | shipped |
 | D | Body-language detection (whatlanggo) + translation cache + async worker | planned |
 
 ## Tables
@@ -40,14 +40,17 @@ Three Postgres tables in the `backend` schema:
 | Action | Caller | Pre-conditions |
 |--------|--------|----------------|
 | Send personal | user | active membership in game; recipient is active member |
+| Paid-tier broadcast | paid-tier user | active membership; recipients = every other active member |
 | Send admin (single user) | game owner OR site admin | recipient is any-status member of the game |
 | Send admin (broadcast) | game owner OR site admin | recipient scope ∈ `active` / `active_and_removed` / `all_members`; sender excluded |
+| Multi-game admin broadcast | site admin | scope `selected` (with `game_ids`) or `all_running` |
+| Bulk purge | site admin | `older_than_years >= 1`; targets games with terminal status finished more than N years ago |
 | Read message | the recipient | row exists in `diplomail_recipients(message_id, user_id)`; non-active members see admin-kind only |
 | Mark read | the recipient | row exists; idempotent if already marked |
 | Soft delete | the recipient | `read_at IS NOT NULL` (open-then-delete, item 10) |
 
-Stage C will add the paid-tier player broadcast and the bulk-purge
-admin endpoint.
+Stage D will add body-language detection (whatlanggo) and the
+translation cache + async worker.
 
 System mail is produced internally by lobby lifecycle hooks:
 `Service.transition()` emits `game.paused` / `game.cancelled` system

backend/internal/diplomail/admin_send.go

@@ -103,6 +103,225 @@ func (s *Service) SendAdminBroadcast(ctx context.Context, in SendAdminBroadcastI
 	return msg, recipients, nil
 }
 
+// SendPlayerBroadcast persists a paid-tier player broadcast and
+// fans out the push event to every other active member of the game.
+// The send is `kind="personal"`, `sender_kind="player"`,
+// `broadcast_scope="game_broadcast"` — recipients reply to it as if
+// it were a single-recipient personal send, and the reply targets
+// only the broadcaster. The caller's entitlement tier is checked
+// against `EntitlementReader`; free-tier callers are rejected with
+// ErrForbidden.
+func (s *Service) SendPlayerBroadcast(ctx context.Context, in SendPlayerBroadcastInput) (Message, []Recipient, error) {
+	subject, body, err := s.prepareContent(in.Subject, in.Body)
+	if err != nil {
+		return Message{}, nil, err
+	}
+	if s.deps.Entitlements == nil {
+		return Message{}, nil, fmt.Errorf("%w: entitlement reader is not wired", ErrForbidden)
+	}
+	paid, err := s.deps.Entitlements.IsPaidTier(ctx, in.SenderUserID)
+	if err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail: entitlement lookup: %w", err)
+	}
+	if !paid {
+		return Message{}, nil, fmt.Errorf("%w: in-game broadcast requires a paid tier", ErrForbidden)
+	}
+
+	sender, err := s.deps.Memberships.GetActiveMembership(ctx, in.GameID, in.SenderUserID)
+	if err != nil {
+		if errors.Is(err, ErrNotFound) {
+			return Message{}, nil, fmt.Errorf("%w: sender is not an active member of the game", ErrForbidden)
+		}
+		return Message{}, nil, fmt.Errorf("diplomail: load sender membership: %w", err)
+	}
+
+	members, err := s.deps.Memberships.ListMembers(ctx, in.GameID, RecipientScopeActive)
+	if err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail: list active members: %w", err)
+	}
+	callerID := in.SenderUserID
+	members = filterOutCaller(members, &callerID)
+	if len(members) == 0 {
+		return Message{}, nil, fmt.Errorf("%w: no other active members in this game", ErrInvalidInput)
+	}
+
+	username := sender.UserName
+	msgInsert := MessageInsert{
+		MessageID:      uuid.New(),
+		GameID:         in.GameID,
+		GameName:       sender.GameName,
+		Kind:           KindPersonal,
+		SenderKind:     SenderKindPlayer,
+		SenderUserID:   &callerID,
+		SenderUsername: &username,
+		SenderIP:       in.SenderIP,
+		Subject:        subject,
+		Body:           body,
+		BodyLang:       LangUndetermined,
+		BroadcastScope: BroadcastScopeGameBroadcast,
+	}
+	rcptInserts := make([]RecipientInsert, 0, len(members))
+	for _, m := range members {
+		rcptInserts = append(rcptInserts, buildRecipientInsert(msgInsert.MessageID, m))
+	}
+	msg, recipients, err := s.deps.Store.InsertMessageWithRecipients(ctx, msgInsert, rcptInserts)
+	if err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail: send player broadcast: %w", err)
+	}
+	for _, r := range recipients {
+		s.publishMessageReceived(ctx, msg, r)
+	}
+	return msg, recipients, nil
+}
+
+// SendAdminMultiGameBroadcast emits one admin-kind message per game
+// resolved from the input scope and fans out the push events. A
+// recipient who plays in multiple addressed games receives one
+// independently-deletable inbox entry per game; this avoids cross-
+// game leakage of admin context and keeps the per-game unread badge
+// honest.
+func (s *Service) SendAdminMultiGameBroadcast(ctx context.Context, in SendMultiGameBroadcastInput) ([]Message, int, error) {
+	subject, body, err := s.prepareContent(in.Subject, in.Body)
+	if err != nil {
+		return nil, 0, err
+	}
+	if err := validateCaller(CallerKindAdmin, nil, in.CallerUsername); err != nil {
+		return nil, 0, err
+	}
+	scope, err := normaliseScope(in.RecipientScope)
+	if err != nil {
+		return nil, 0, err
+	}
+	if s.deps.Games == nil {
+		return nil, 0, fmt.Errorf("%w: game lookup is not wired", ErrInvalidInput)
+	}
+	games, err := s.resolveMultiGameTargets(ctx, in)
+	if err != nil {
+		return nil, 0, err
+	}
+	if len(games) == 0 {
+		return nil, 0, fmt.Errorf("%w: no games match the broadcast scope", ErrInvalidInput)
+	}
+
+	totalRecipients := 0
+	out := make([]Message, 0, len(games))
+	for _, game := range games {
+		members, err := s.deps.Memberships.ListMembers(ctx, game.GameID, scope)
+		if err != nil {
+			return nil, 0, fmt.Errorf("diplomail: list members for %s: %w", game.GameID, err)
+		}
+		if len(members) == 0 {
+			s.deps.Logger.Debug("multi-game broadcast skips empty game",
+				zap.String("game_id", game.GameID.String()),
+				zap.String("scope", scope))
+			continue
+		}
+		msgInsert, err := s.buildAdminMessageInsert(CallerKindAdmin, nil, in.CallerUsername,
+			game.GameID, game.GameName, subject, body, in.SenderIP, BroadcastScopeMultiGameBroadcast)
+		if err != nil {
+			return nil, 0, err
+		}
+		rcptInserts := make([]RecipientInsert, 0, len(members))
+		for _, m := range members {
+			rcptInserts = append(rcptInserts, buildRecipientInsert(msgInsert.MessageID, m))
+		}
+		msg, recipients, err := s.deps.Store.InsertMessageWithRecipients(ctx, msgInsert, rcptInserts)
+		if err != nil {
+			return nil, 0, fmt.Errorf("diplomail: insert multi-game broadcast for %s: %w", game.GameID, err)
+		}
+		for _, r := range recipients {
+			s.publishMessageReceived(ctx, msg, r)
+		}
+		out = append(out, msg)
+		totalRecipients += len(recipients)
+	}
+	return out, totalRecipients, nil
+}
+
+func (s *Service) resolveMultiGameTargets(ctx context.Context, in SendMultiGameBroadcastInput) ([]GameSnapshot, error) {
+	switch in.Scope {
+	case MultiGameScopeAllRunning:
+		games, err := s.deps.Games.ListRunningGames(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("diplomail: list running games: %w", err)
+		}
+		return games, nil
+	case MultiGameScopeSelected, "":
+		if len(in.GameIDs) == 0 {
+			return nil, fmt.Errorf("%w: selected scope requires game_ids", ErrInvalidInput)
+		}
+		out := make([]GameSnapshot, 0, len(in.GameIDs))
+		for _, id := range in.GameIDs {
+			game, err := s.deps.Games.GetGame(ctx, id)
+			if err != nil {
+				if errors.Is(err, ErrNotFound) {
+					return nil, fmt.Errorf("%w: game %s not found", ErrInvalidInput, id)
+				}
+				return nil, fmt.Errorf("diplomail: load game %s: %w", id, err)
+			}
+			out = append(out, game)
+		}
+		return out, nil
+	default:
+		return nil, fmt.Errorf("%w: unknown multi-game scope %q", ErrInvalidInput, in.Scope)
+	}
+}
+
+// BulkCleanup deletes every diplomail_messages row tied to games that
+// finished more than `OlderThanYears` years ago. Returns the affected
+// game ids and the count of removed messages. The minimum allowed
+// value is 1 year — finer-grained pruning would risk wiping live
+// arbitration evidence.
+func (s *Service) BulkCleanup(ctx context.Context, in BulkCleanupInput) (CleanupResult, error) {
+	if in.OlderThanYears < 1 {
+		return CleanupResult{}, fmt.Errorf("%w: older_than_years must be >= 1", ErrInvalidInput)
+	}
+	if s.deps.Games == nil {
+		return CleanupResult{}, fmt.Errorf("%w: game lookup is not wired", ErrInvalidInput)
+	}
+	cutoff := s.nowUTC().AddDate(-in.OlderThanYears, 0, 0)
+	games, err := s.deps.Games.ListFinishedGamesBefore(ctx, cutoff)
+	if err != nil {
+		return CleanupResult{}, fmt.Errorf("diplomail: list finished games: %w", err)
+	}
+	if len(games) == 0 {
+		return CleanupResult{}, nil
+	}
+	gameIDs := make([]uuid.UUID, 0, len(games))
+	for _, g := range games {
+		gameIDs = append(gameIDs, g.GameID)
+	}
+	deleted, err := s.deps.Store.DeleteMessagesForGames(ctx, gameIDs)
+	if err != nil {
+		return CleanupResult{}, fmt.Errorf("diplomail: bulk delete: %w", err)
+	}
+	return CleanupResult{GameIDs: gameIDs, MessagesDeleted: deleted}, nil
+}
+
+// ListMessagesForAdmin returns a paginated, optionally-filtered view
+// of every persisted message. Used by the admin observability
+// endpoint to inspect what has been sent and trace abuse reports.
+func (s *Service) ListMessagesForAdmin(ctx context.Context, filter AdminMessageListing) (AdminMessagePage, error) {
+	rows, total, err := s.deps.Store.ListMessagesForAdmin(ctx, filter)
+	if err != nil {
+		return AdminMessagePage{}, err
+	}
+	page := filter.Page
+	if page < 1 {
+		page = 1
+	}
+	pageSize := filter.PageSize
+	if pageSize < 1 {
+		pageSize = 50
+	}
+	return AdminMessagePage{
+		Items:    rows,
+		Total:    total,
+		Page:     page,
+		PageSize: pageSize,
+	}, nil
+}
+
 // PublishLifecycle persists a system-kind message in response to a
 // lobby lifecycle transition and fan-outs push events to the
 // affected recipients. Game-scoped transitions (`game.paused`,

backend/internal/diplomail/deps.go

@@ -15,16 +15,61 @@ import (
 // Store and Memberships are required. Logger and Now default to
 // zap.NewNop / time.Now when nil. Notification falls back to a no-op
 // publisher so unit tests can construct a Service with only the
-// required collaborators populated.
+// required collaborators populated. Entitlements and Games are
+// optional — they are used by Stage C surfaces (paid-tier player
+// broadcast, multi-game admin broadcast, bulk cleanup). Wiring may
+// pass nil for tests that do not exercise those paths.
 type Deps struct {
 	Store        *Store
 	Memberships  MembershipLookup
 	Notification NotificationPublisher
+	Entitlements EntitlementReader
+	Games        GameLookup
 	Config       config.DiplomailConfig
 	Logger       *zap.Logger
 	Now          func() time.Time
 }
 
+// EntitlementReader is the read-only surface diplomail uses to gate
+// the paid-tier player broadcast. The canonical implementation in
+// `cmd/backend/main` reads
+// `*user.Service.GetEntitlementSnapshot(userID).IsPaid`.
+type EntitlementReader interface {
+	IsPaidTier(ctx context.Context, userID uuid.UUID) (bool, error)
+}
+
+// GameLookup exposes the slim view of `games` the multi-game admin
+// broadcast and bulk-cleanup paths consume. The canonical
+// implementation walks the lobby cache plus an explicit store call
+// for finished-game pruning.
+type GameLookup interface {
+	// ListRunningGames returns every game whose `status` is one of
+	// the still-active values (running, paused, starting, …). The
+	// admin `all_running` broadcast scope iterates over the result.
+	ListRunningGames(ctx context.Context) ([]GameSnapshot, error)
+
+	// ListFinishedGamesBefore returns every game whose `finished_at`
+	// is older than `cutoff`. The bulk-purge admin endpoint reads
+	// this to compose the cascade-delete IN list.
+	ListFinishedGamesBefore(ctx context.Context, cutoff time.Time) ([]GameSnapshot, error)
+
+	// GetGame returns one game snapshot identified by id, or
+	// ErrNotFound. Used by the multi-game broadcast to verify the
+	// caller-supplied id list before enqueuing fan-out work.
+	GetGame(ctx context.Context, gameID uuid.UUID) (GameSnapshot, error)
+}
+
+// GameSnapshot is the trim view of `games` consumed by the multi-game
+// admin broadcast and the cleanup paths. The struct intentionally
+// avoids the full `lobby.GameRecord` so the diplomail package stays
+// decoupled from the lobby domain.
+type GameSnapshot struct {
+	GameID      uuid.UUID
+	GameName    string
+	Status      string
+	FinishedAt  *time.Time
+}
+
 // ActiveMembership is the slim view of a single (user, game) roster
 // row the diplomail package needs at send time: it confirms the
 // participant is active in the game and captures the snapshot fields

backend/internal/diplomail/diplomail_e2e_test.go

@@ -510,6 +510,249 @@ func TestDiplomailAdminBroadcast(t *testing.T) {
 	}
 }
 
+// staticEntitlement satisfies diplomail.EntitlementReader by reading
+// a fixed map keyed on user_id.
+type staticEntitlement struct {
+	paid map[uuid.UUID]bool
+}
+
+func (s *staticEntitlement) IsPaidTier(_ context.Context, userID uuid.UUID) (bool, error) {
+	if s == nil {
+		return false, nil
+	}
+	return s.paid[userID], nil
+}
+
+// staticGameLookup satisfies diplomail.GameLookup by walking a fixed
+// list of GameSnapshot fixtures. Tests prepend rows via the New
+// helper.
+type staticGameLookup struct {
+	games map[uuid.UUID]diplomail.GameSnapshot
+}
+
+func (l *staticGameLookup) ListRunningGames(_ context.Context) ([]diplomail.GameSnapshot, error) {
+	if l == nil {
+		return nil, nil
+	}
+	out := make([]diplomail.GameSnapshot, 0, len(l.games))
+	for _, g := range l.games {
+		switch g.Status {
+		case "running", "paused", "ready_to_start", "starting":
+			out = append(out, g)
+		}
+	}
+	return out, nil
+}
+
+func (l *staticGameLookup) ListFinishedGamesBefore(_ context.Context, cutoff time.Time) ([]diplomail.GameSnapshot, error) {
+	if l == nil {
+		return nil, nil
+	}
+	out := make([]diplomail.GameSnapshot, 0, len(l.games))
+	for _, g := range l.games {
+		if g.Status != "finished" && g.Status != "cancelled" {
+			continue
+		}
+		if g.FinishedAt == nil || !g.FinishedAt.Before(cutoff) {
+			continue
+		}
+		out = append(out, g)
+	}
+	return out, nil
+}
+
+func (l *staticGameLookup) GetGame(_ context.Context, gameID uuid.UUID) (diplomail.GameSnapshot, error) {
+	if l == nil {
+		return diplomail.GameSnapshot{}, diplomail.ErrNotFound
+	}
+	g, ok := l.games[gameID]
+	if !ok {
+		return diplomail.GameSnapshot{}, diplomail.ErrNotFound
+	}
+	return g, nil
+}
+
+func TestDiplomailPaidTierBroadcast(t *testing.T) {
+	db := startPostgres(t)
+	ctx := context.Background()
+
+	gameID := uuid.New()
+	paidPlayer := uuid.New()
+	freePlayer := uuid.New()
+	alice := uuid.New()
+	bob := uuid.New()
+	seedAccount(t, db, paidPlayer)
+	seedAccount(t, db, freePlayer)
+	seedAccount(t, db, alice)
+	seedAccount(t, db, bob)
+	seedGame(t, db, gameID, "Paid Broadcast Game")
+
+	lookup := &staticMembershipLookup{
+		rows: map[[2]uuid.UUID]diplomail.ActiveMembership{
+			{gameID, paidPlayer}: {
+				UserID: paidPlayer, GameID: gameID, GameName: "Paid Broadcast Game",
+				UserName: "paid", RaceName: "PaidRace",
+			},
+			{gameID, freePlayer}: {
+				UserID: freePlayer, GameID: gameID, GameName: "Paid Broadcast Game",
+				UserName: "free", RaceName: "FreeRace",
+			},
+			{gameID, alice}: {
+				UserID: alice, GameID: gameID, GameName: "Paid Broadcast Game",
+				UserName: "alice", RaceName: "AliceRace",
+			},
+			{gameID, bob}: {
+				UserID: bob, GameID: gameID, GameName: "Paid Broadcast Game",
+				UserName: "bob", RaceName: "BobRace",
+			},
+		},
+	}
+	publisher := &recordingPublisher{}
+	entitlements := &staticEntitlement{paid: map[uuid.UUID]bool{paidPlayer: true}}
+
+	svc := diplomail.NewService(diplomail.Deps{
+		Store:        diplomail.NewStore(db),
+		Memberships:  lookup,
+		Notification: publisher,
+		Entitlements: entitlements,
+		Config: config.DiplomailConfig{
+			MaxBodyBytes:    4096,
+			MaxSubjectBytes: 256,
+		},
+	})
+
+	// Paid sender: broadcast succeeds.
+	msg, recipients, err := svc.SendPlayerBroadcast(ctx, diplomail.SendPlayerBroadcastInput{
+		GameID:       gameID,
+		SenderUserID: paidPlayer,
+		Subject:      "Alliance",
+		Body:         "Let us form a coalition.",
+	})
+	if err != nil {
+		t.Fatalf("paid broadcast: %v", err)
+	}
+	if msg.Kind != diplomail.KindPersonal || msg.BroadcastScope != diplomail.BroadcastScopeGameBroadcast {
+		t.Fatalf("kind=%q scope=%q, want personal/game_broadcast", msg.Kind, msg.BroadcastScope)
+	}
+	if len(recipients) != 3 {
+		t.Fatalf("broadcast recipients=%d, want 3 (everyone but sender)", len(recipients))
+	}
+
+	// Free-tier sender: 403.
+	_, _, err = svc.SendPlayerBroadcast(ctx, diplomail.SendPlayerBroadcastInput{
+		GameID:       gameID,
+		SenderUserID: freePlayer,
+		Body:         "Should not be allowed.",
+	})
+	if !errors.Is(err, diplomail.ErrForbidden) {
+		t.Fatalf("free broadcast: %v, want ErrForbidden", err)
+	}
+}
+
+func TestDiplomailMultiGameBroadcastAndCleanup(t *testing.T) {
+	db := startPostgres(t)
+	ctx := context.Background()
+
+	game1 := uuid.New()
+	game2 := uuid.New()
+	finished := uuid.New()
+	alice := uuid.New()
+	bob := uuid.New()
+	carol := uuid.New()
+	for _, id := range []uuid.UUID{alice, bob, carol} {
+		seedAccount(t, db, id)
+	}
+	seedGame(t, db, game1, "Active Game 1")
+	seedGame(t, db, game2, "Active Game 2")
+	seedGame(t, db, finished, "Finished Game")
+	// Mark `finished` terminal with a long-past finished_at.
+	if _, err := db.ExecContext(ctx, `
+		UPDATE backend.games
+		   SET status='finished', finished_at = now() - interval '3 years'
+		 WHERE game_id = $1
+	`, finished); err != nil {
+		t.Fatalf("backdate finished: %v", err)
+	}
+
+	lookup := &staticMembershipLookup{
+		rows: map[[2]uuid.UUID]diplomail.ActiveMembership{
+			{game1, alice}: {UserID: alice, GameID: game1, GameName: "Active Game 1", UserName: "alice", RaceName: "AliceRace"},
+			{game1, bob}:   {UserID: bob, GameID: game1, GameName: "Active Game 1", UserName: "bob", RaceName: "BobRace"},
+			{game2, carol}: {UserID: carol, GameID: game2, GameName: "Active Game 2", UserName: "carol", RaceName: "CarolRace"},
+			{finished, alice}: {UserID: alice, GameID: finished, GameName: "Finished Game", UserName: "alice", RaceName: "AliceRace"},
+		},
+	}
+	publisher := &recordingPublisher{}
+	finAt := time.Now().UTC().AddDate(-3, 0, 0)
+	games := &staticGameLookup{games: map[uuid.UUID]diplomail.GameSnapshot{
+		game1:    {GameID: game1, GameName: "Active Game 1", Status: "running"},
+		game2:    {GameID: game2, GameName: "Active Game 2", Status: "running"},
+		finished: {GameID: finished, GameName: "Finished Game", Status: "finished", FinishedAt: &finAt},
+	}}
+
+	svc := diplomail.NewService(diplomail.Deps{
+		Store:        diplomail.NewStore(db),
+		Memberships:  lookup,
+		Notification: publisher,
+		Games:        games,
+		Config: config.DiplomailConfig{
+			MaxBodyBytes:    4096,
+			MaxSubjectBytes: 256,
+		},
+	})
+
+	// First, drop a personal message into the finished game so cleanup
+	// has something to remove.
+	if _, _, err := svc.SendAdminPersonal(ctx, diplomail.SendAdminPersonalInput{
+		GameID:          finished,
+		CallerKind:      diplomail.CallerKindAdmin,
+		CallerUsername:  "ops",
+		RecipientUserID: alice,
+		Body:            "Audit ping",
+	}); err != nil {
+		t.Fatalf("seed finished-game mail: %v", err)
+	}
+
+	// Multi-game broadcast across all running games.
+	msgs, total, err := svc.SendAdminMultiGameBroadcast(ctx, diplomail.SendMultiGameBroadcastInput{
+		CallerUsername: "ops",
+		Scope:          diplomail.MultiGameScopeAllRunning,
+		RecipientScope: diplomail.RecipientScopeActive,
+		Subject:        "Maintenance",
+		Body:           "Brief turn-engine restart in 10 minutes.",
+	})
+	if err != nil {
+		t.Fatalf("multi-game broadcast: %v", err)
+	}
+	if len(msgs) != 2 {
+		t.Fatalf("multi-game messages=%d, want 2 (game1 + game2)", len(msgs))
+	}
+	if total != 3 {
+		t.Fatalf("multi-game recipient count=%d, want 3 (alice+bob in g1, carol in g2)", total)
+	}
+
+	// Bulk cleanup with 1-year cutoff should sweep the finished game.
+	result, err := svc.BulkCleanup(ctx, diplomail.BulkCleanupInput{OlderThanYears: 1})
+	if err != nil {
+		t.Fatalf("bulk cleanup: %v", err)
+	}
+	if len(result.GameIDs) != 1 || result.GameIDs[0] != finished {
+		t.Fatalf("cleanup game_ids=%v, want [%s]", result.GameIDs, finished)
+	}
+	if result.MessagesDeleted < 1 {
+		t.Fatalf("cleanup messages_deleted=%d, want >=1", result.MessagesDeleted)
+	}
+
+	// Admin listing sees the multi-game messages.
+	page, err := svc.ListMessagesForAdmin(ctx, diplomail.AdminMessageListing{Page: 1, PageSize: 50})
+	if err != nil {
+		t.Fatalf("list admin messages: %v", err)
+	}
+	if page.Total < 2 {
+		t.Fatalf("list total=%d, want >=2 after cleanup", page.Total)
+	}
+}
+
 func TestDiplomailLifecycleMembershipKick(t *testing.T) {
 	db := startPostgres(t)
 	ctx := context.Background()

backend/internal/diplomail/store.go

@@ -358,6 +358,99 @@ func (s *Store) UnreadCountForUserGame(ctx context.Context, gameID, userID uuid.
 	return int(dest.Count), nil
 }
 
+// DeleteMessagesForGames removes every diplomail_messages row whose
+// game_id falls in the supplied set. The cascade defined on the
+// `diplomail_recipients` and `diplomail_translations` foreign keys
+// removes the per-recipient state and the cached translations in
+// the same transaction. Returns the count of messages removed.
+//
+// Used by the admin bulk-purge endpoint; callers are expected to
+// have already filtered the input set to terminal-state games.
+func (s *Store) DeleteMessagesForGames(ctx context.Context, gameIDs []uuid.UUID) (int, error) {
+	if len(gameIDs) == 0 {
+		return 0, nil
+	}
+	args := make([]postgres.Expression, 0, len(gameIDs))
+	for _, id := range gameIDs {
+		args = append(args, postgres.UUID(id))
+	}
+	m := table.DiplomailMessages
+	stmt := m.DELETE().WHERE(m.GameID.IN(args...))
+	res, err := stmt.ExecContext(ctx, s.db)
+	if err != nil {
+		return 0, fmt.Errorf("diplomail store: bulk delete messages: %w", err)
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("diplomail store: rows affected: %w", err)
+	}
+	return int(affected), nil
+}
+
+// ListMessagesForAdmin returns a paginated slice of messages
+// matching filter. The result is ordered by created_at DESC,
+// message_id DESC. Total is the count without pagination so the
+// caller can render a "page X of N" envelope.
+func (s *Store) ListMessagesForAdmin(ctx context.Context, filter AdminMessageListing) ([]Message, int, error) {
+	m := table.DiplomailMessages
+	page := filter.Page
+	if page < 1 {
+		page = 1
+	}
+	pageSize := filter.PageSize
+	if pageSize < 1 {
+		pageSize = 50
+	}
+
+	conditions := postgres.BoolExpression(nil)
+	addCondition := func(cond postgres.BoolExpression) {
+		if conditions == nil {
+			conditions = cond
+			return
+		}
+		conditions = conditions.AND(cond)
+	}
+	if filter.GameID != nil {
+		addCondition(m.GameID.EQ(postgres.UUID(*filter.GameID)))
+	}
+	if filter.Kind != "" {
+		addCondition(m.Kind.EQ(postgres.String(filter.Kind)))
+	}
+	if filter.SenderKind != "" {
+		addCondition(m.SenderKind.EQ(postgres.String(filter.SenderKind)))
+	}
+
+	countStmt := postgres.SELECT(postgres.COUNT(postgres.STAR).AS("count")).FROM(m)
+	if conditions != nil {
+		countStmt = countStmt.WHERE(conditions)
+	}
+	var countDest struct {
+		Count int64 `alias:"count"`
+	}
+	if err := countStmt.QueryContext(ctx, s.db, &countDest); err != nil {
+		return nil, 0, fmt.Errorf("diplomail store: count admin messages: %w", err)
+	}
+
+	listStmt := postgres.SELECT(messageColumns()).FROM(m)
+	if conditions != nil {
+		listStmt = listStmt.WHERE(conditions)
+	}
+	listStmt = listStmt.
+		ORDER_BY(m.CreatedAt.DESC(), m.MessageID.DESC()).
+		LIMIT(int64(pageSize)).
+		OFFSET(int64((page - 1) * pageSize))
+
+	var rows []model.DiplomailMessages
+	if err := listStmt.QueryContext(ctx, s.db, &rows); err != nil {
+		return nil, 0, fmt.Errorf("diplomail store: list admin messages: %w", err)
+	}
+	out := make([]Message, 0, len(rows))
+	for _, row := range rows {
+		out = append(out, messageFromModel(row))
+	}
+	return out, int(countDest.Count), nil
+}
+
 // UnreadCountsForUser returns a per-game breakdown of unread messages
 // addressed to userID, plus the matching game names so the lobby
 // badge UI can render entries even after the recipient's membership

backend/internal/diplomail/types.go

@@ -120,6 +120,78 @@ const (
 	LifecycleKindMembershipBlocked   = "membership.blocked"
 )
 
+// SendPlayerBroadcastInput is the request payload for the paid-tier
+// player broadcast. The sender is a player; recipients are the
+// active members of the game minus the sender. The resulting message
+// is `kind="personal"`, `sender_kind="player"`,
+// `broadcast_scope="game_broadcast"` — recipients may reply as if it
+// were a personal send, but the reply goes back to the broadcaster
+// only.
+type SendPlayerBroadcastInput struct {
+	GameID       uuid.UUID
+	SenderUserID uuid.UUID
+	Subject      string
+	Body         string
+	SenderIP     string
+}
+
+// MultiGameBroadcastScope enumerates the admin multi-game broadcast
+// modes. `selected` requires `GameIDs`; `all_running` enumerates
+// every game whose status is non-terminal through GameLookup.
+const (
+	MultiGameScopeSelected   = "selected"
+	MultiGameScopeAllRunning = "all_running"
+)
+
+// SendMultiGameBroadcastInput is the request payload for the admin
+// multi-game broadcast. The service materialises one message row per
+// addressed game (so a recipient who plays in two games receives two
+// independently-deletable inbox entries), then fan-outs the push
+// events.
+type SendMultiGameBroadcastInput struct {
+	CallerUsername string
+	Scope          string
+	GameIDs        []uuid.UUID
+	RecipientScope string
+	Subject        string
+	Body           string
+	SenderIP       string
+}
+
+// BulkCleanupInput selects messages eligible for purge. OlderThanYears
+// must be >= 1; the service translates the value into a cutoff
+// expressed in years and walks `GameLookup.ListFinishedGamesBefore`.
+type BulkCleanupInput struct {
+	OlderThanYears int
+}
+
+// CleanupResult summarises a bulk-cleanup run for the admin response
+// envelope.
+type CleanupResult struct {
+	GameIDs        []uuid.UUID
+	MessagesDeleted int
+}
+
+// AdminMessageListing is the filter passed to ListMessagesForAdmin.
+// Pagination uses (Page, PageSize) consistent with the rest of the
+// admin surface. Filters are AND-combined; the empty filter returns
+// every persisted row.
+type AdminMessageListing struct {
+	Page     int
+	PageSize int
+	GameID   *uuid.UUID
+	Kind     string
+	SenderKind string
+}
+
+// AdminMessagePage is the canonical pagination envelope.
+type AdminMessagePage struct {
+	Items    []Message
+	Total    int
+	Page     int
+	PageSize int
+}
+
 // LifecycleEvent is the payload lobby hands to PublishLifecycle when
 // a transition needs to be reflected as durable system mail. The
 // recipient set is derived by the service: