feat(mail): add broadcast mail for admins

Adds endpoints for PLATFORM_ADMIN to send mail to all active users or
picked ones. Broadcasts and account mails share the same outbox;
broadcasts take the rest of the daily limit after 10 mails are kept free
for account mail. Users can opt out via settings. History is kept in
mail_broadcast with recipient, sent and failed counts. The footer says
where to opt out.

readme.md

@@ -86,6 +86,13 @@ To Be Played is split into three repositories:
   to the admin page), and to the user "we got your registration", then "your account is active" or "your
   registration was denied", later "your account is blocked" and "your account will be deleted from <date>". The registration sends the browser language (`de` for a German browser, else `en`), which also becomes
   the user's app language. Plain text only. See "Mail flow".
+- **Broadcast mail**: a PLATFORM_ADMIN writes one mail with subject and text in English and German (plain text),
+  and sends it to all active users or to picked ones, each in their app language. A user can turn it off
+  (`user_settings.platform_mails`, default on); the footer of the mail says where. "Send test to me" mails the draft
+  to the admin only, in the admin's language. One `mail_broadcast` row keeps the texts and the counts (recipients,
+  sent, failed); `BROADCAST_SEND` goes to `admin_audit`. The sender takes account mails first and stays under
+  `APP_MAIL_DAILY_LIMIT` per 24 h, 10 of them kept for account mails. A recipient who is blocked, deleted or opted
+  out before the send counts as failed.
 - **Delete my account** with the password (5 wrong tries per 15 minutes, then `429`). It only sets
   `users.deleted_at` and revokes every refresh token: from then on no login, no refresh, no access token, no public
   profile, no Steam link, not in the landing stats. Email and display name stay taken; the Steam ID is freed, so an
@@ -128,6 +135,11 @@ To Be Played is split into three repositories:
 | `PUT` | `/api/users/settings/profile-style` (`{accent, corners, bannerAppid, backgroundAppid}`; `204`) | access token |
 | `PUT` | `/api/users/settings/dashboard-tiles` (`{tiles: [{id, size}]}` in order, at most 20; `{tiles: null}` resets; `204`). Auth checks only the id shape `[A-Z_]{1,30}`: Core and the Frontend skip ids they do not know, and Core never shares the overview or the activity | access token |
 | `PUT` | `/api/users/settings/welcome-seen` (`204`; stamps `user_settings.welcome_seen_at`, sent as `settings.welcomeSeenAt` in `/me`, null until then) | access token |
+| `PUT` | `/api/users/settings/platform-mails` (`{enabled}`; `204`) | access token |
+| `GET` | `/api/admin/broadcasts` (`{results}` with the counts, newest first) | PLATFORM_ADMIN |
+| `GET` | `/api/admin/broadcasts/recipient-count?userIds=` (`{count}`; no `userIds` = all) | PLATFORM_ADMIN |
+| `POST` | `/api/admin/broadcasts` (`{subjectEn, subjectDe, bodyEn, bodyDe, userIds}`; `201 {recipientCount}`; `409 NO_RECIPIENTS`) | PLATFORM_ADMIN |
+| `POST` | `/api/admin/broadcasts/test` (same body; `204`) | PLATFORM_ADMIN |
 | `PUT` | `/api/users/settings/favorite-platform` (`{platform}`, free text up to 32, e.g. `PS5`; `null` clears it; `204`) | access token |
 | `PUT` | `/api/users/settings/profile-tiles` (`{tiles: [{id, size}]}`, e.g. `SUMMARY`, `STATUS`, `PLAYING`, `FINISHED`, `STATS_<GROUP>`; `204`) | access token |
 | `GET` | `/api/public/profiles/{displayName}` | public |
@@ -179,6 +191,8 @@ use each other.
 `mail` writes each mail as a finished text into `mail_outbox` (`JdbcTemplate`, `repo`), in the transaction of the
 event, and `MailOutboxService` sends the rows every minute over SMTP. Other modules reach it through the port
 `shared/mail/AccountMailer`. Without `SPRING_MAIL_HOST` mail is off: the rows wait and go out once a host is set.
+Broadcasts (`BroadcastService`, `mail_broadcast`) queue one row per recipient; the same job sends them after the
+account mails and under the daily limit.
 
 ArchUnit enforces these rules in `src/test/java/auth_service/architecture/ArchitectureTest.java`, so a layer
 violation fails the build. It also keeps the modules independent of each other and free of cycles. Test base
@@ -277,10 +291,12 @@ sequenceDiagram
     participant B as Browser (Frontend)
     participant A as Auth
     participant M as SMTP (Hostinger)
+    B->>A: admin sends a broadcast
+    A->>A: mail_broadcast row + one outbox row per active, opted-in user
     B->>A: register / admin decides / admin blocks / delete account
     A->>A: event + mail_outbox row (PENDING), one transaction
     loop every minute (MailOutboxService)
-        A->>M: up to 20 due rows
+        A->>M: up to 20 due rows (account mails first, at most APP_MAIL_DAILY_LIMIT per 24 h, broadcast rows keep 10 free)
         M-->>A: ok: SENT / error: try again after 1, 2, 4 ... 60 min
     end
     Note over A: after 10 tries FAILED and a warning in the log; SENT rows are removed after 30 days
@@ -381,6 +397,7 @@ Configuration comes from environment variables:
 | `SPRING_MAIL_HOST`, `SPRING_MAIL_USERNAME`, `SPRING_MAIL_PASSWORD` | SMTP server and its login (port 465, SSL). No host = mail off |
 | `APP_MAIL_FROM` | Sender address, the mailbox of the SMTP login. Needed with a host |
 | `APP_MAIL_ADMIN_TO` | Gets "a new registration waits". Needed with a host |
+| `APP_MAIL_DAILY_LIMIT` | Mails per rolling 24 h, all kinds together (default 100, Hostinger's free mailbox; 1000 Business Starter, 3000 Business Standard / Premium). Broadcast rows keep 10 free |
 
 Generate a keypair:
 

src/main/java/auth_service/modules/mail/api/AdminBroadcastController.java

@@ -0,0 +1,69 @@
+package auth_service.modules.mail.api;
+
+import auth_service.modules.mail.api.dto.request.BroadcastRequest;
+import auth_service.modules.mail.api.dto.response.BroadcastListResponse;
+import auth_service.modules.mail.api.dto.response.BroadcastResponse;
+import auth_service.modules.mail.api.dto.response.RecipientCountResponse;
+import auth_service.modules.mail.api.dto.response.SentBroadcastResponse;
+import auth_service.modules.mail.model.Broadcast;
+import auth_service.modules.mail.model.BroadcastText;
+import auth_service.modules.mail.service.BroadcastService;
+import auth_service.shared.security.UserPrincipal;
+import jakarta.validation.Valid;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.core.annotation.AuthenticationPrincipal;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/api/admin/broadcasts")
+@RequiredArgsConstructor
+public class AdminBroadcastController {
+
+    private final BroadcastService broadcastService;
+
+    @GetMapping
+    public ResponseEntity<BroadcastListResponse> findAll() {
+        return ResponseEntity.ok(new BroadcastListResponse(broadcastService.findAll().stream()
+                .map(AdminBroadcastController::toApi).toList()));
+    }
+
+    @GetMapping("/recipient-count")
+    public ResponseEntity<RecipientCountResponse> recipientCount(@RequestParam(required = false) List<Long> userIds) {
+        return ResponseEntity.ok(new RecipientCountResponse(broadcastService.countRecipients(userIds)));
+    }
+
+    @PostMapping
+    public ResponseEntity<SentBroadcastResponse> send(@AuthenticationPrincipal UserPrincipal principal,
+                                                      @Valid @RequestBody BroadcastRequest request) {
+        int queued = broadcastService.send(principal.getIdAsLong(), toText(request), request.userIds());
+        return ResponseEntity.status(HttpStatus.CREATED).body(new SentBroadcastResponse(queued));
+    }
+
+    @PostMapping("/test")
+    public ResponseEntity<Void> sendTest(@AuthenticationPrincipal UserPrincipal principal,
+                                         @Valid @RequestBody BroadcastRequest request) {
+        broadcastService.sendTest(principal.getIdAsLong(), toText(request));
+        return ResponseEntity.noContent().build();
+    }
+
+    private static BroadcastText toText(BroadcastRequest request) {
+        return new BroadcastText(request.subjectEn().strip(), request.subjectDe().strip(),
+                request.bodyEn().strip(), request.bodyDe().strip());
+    }
+
+    private static BroadcastResponse toApi(Broadcast broadcast) {
+        BroadcastText text = broadcast.text();
+        return new BroadcastResponse(broadcast.id(), text.subjectEn(), text.subjectDe(), text.bodyEn(), text.bodyDe(),
+                broadcast.createdBy(), broadcast.createdAt(), broadcast.recipientCount(), broadcast.sentCount(),
+                broadcast.failedCount());
+    }
+}

src/main/java/auth_service/modules/mail/api/BroadcastExceptionController.java

@@ -0,0 +1,21 @@
+package auth_service.modules.mail.api;
+
+import auth_service.modules.mail.exception.NoRecipientsException;
+import auth_service.shared.api.dto.response.ErrorResponse;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.RestControllerAdvice;
+
+@RestControllerAdvice(assignableTypes = AdminBroadcastController.class)
+@Order(value = 1)
+public class BroadcastExceptionController {
+
+    @ExceptionHandler(NoRecipientsException.class)
+    public ResponseEntity<ErrorResponse> conflict(NoRecipientsException ex) {
+        return ResponseEntity.status(HttpStatus.CONFLICT).contentType(MediaType.APPLICATION_JSON)
+                .body(new ErrorResponse("NO_RECIPIENTS", ex.getMessage()));
+    }
+}

src/main/java/auth_service/modules/mail/api/dto/request/BroadcastRequest.java

@@ -0,0 +1,16 @@
+package auth_service.modules.mail.api.dto.request;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Size;
+
+import java.util.List;
+
+/**
+ * Plain text, never HTML.
+ *
+ * @param userIds null = every active user with platform mails on; the test mail ignores it
+ */
+public record BroadcastRequest(@NotBlank @Size(max = 150) String subjectEn, @NotBlank @Size(max = 150) String subjectDe,
+                               @NotBlank @Size(max = 5000) String bodyEn, @NotBlank @Size(max = 5000) String bodyDe,
+                               List<Long> userIds) {
+}

src/main/java/auth_service/modules/mail/api/dto/response/BroadcastListResponse.java

@@ -0,0 +1,6 @@
+package auth_service.modules.mail.api.dto.response;
+
+import java.util.List;
+
+public record BroadcastListResponse(List<BroadcastResponse> results) {
+}

src/main/java/auth_service/modules/mail/api/dto/response/BroadcastResponse.java

@@ -0,0 +1,8 @@
+package auth_service.modules.mail.api.dto.response;
+
+import java.time.Instant;
+
+public record BroadcastResponse(long id, String subjectEn, String subjectDe, String bodyEn, String bodyDe,
+                                long createdBy, Instant createdAt, int recipientCount, int sentCount,
+                                int failedCount) {
+}

src/main/java/auth_service/modules/mail/api/dto/response/RecipientCountResponse.java

@@ -0,0 +1,4 @@
+package auth_service.modules.mail.api.dto.response;
+
+public record RecipientCountResponse(long count) {
+}

src/main/java/auth_service/modules/mail/api/dto/response/SentBroadcastResponse.java

@@ -0,0 +1,4 @@
+package auth_service.modules.mail.api.dto.response;
+
+public record SentBroadcastResponse(int recipientCount) {
+}

src/main/java/auth_service/modules/mail/exception/NoRecipientsException.java

@@ -0,0 +1,8 @@
+package auth_service.modules.mail.exception;
+
+public class NoRecipientsException extends RuntimeException {
+
+    public NoRecipientsException() {
+        super("No active user with platform mails on among the recipients");
+    }
+}

src/main/java/auth_service/modules/mail/model/Broadcast.java

@@ -0,0 +1,7 @@
+package auth_service.modules.mail.model;
+
+import java.time.Instant;
+
+public record Broadcast(long id, BroadcastText text, long createdBy, Instant createdAt, int recipientCount,
+                        int sentCount, int failedCount) {
+}

src/main/java/auth_service/modules/mail/model/BroadcastText.java

@@ -0,0 +1,4 @@
+package auth_service.modules.mail.model;
+
+public record BroadcastText(String subjectEn, String subjectDe, String bodyEn, String bodyDe) {
+}

src/main/java/auth_service/modules/mail/model/MailTemplate.java

@@ -117,6 +117,23 @@ public enum MailTemplate {
             Alle deine Daten werden ab dem %s gelöscht.
 
             ToBePlayed
+            """),
+
+    // Admin text only: %1$s subject, %2$s body, %3$s settings URL.
+    BROADCAST(
+            "%1$s", """
+            %2$s
+
+            --
+            You get this mail because you have an account on ToBePlayed. Turn these mails off in your settings:
+            %3$s
+            """,
+            "%1$s", """
+            %2$s
+
+            --
+            Du bekommst diese Mail, weil du ein Konto bei ToBePlayed hast. In den Einstellungen kannst du sie abschalten:
+            %3$s
             """);
 
     private static final DateTimeFormatter DATE_DE = DateTimeFormatter.ofPattern("dd.MM.yyyy");
@@ -135,14 +152,14 @@ public enum MailTemplate {
      */
     public Mail render(String language, Object... args) {
         Mail mail = isGerman(language) ? de : en;
-        return new Mail(mail.subject(), mail.body().formatted(args));
+        return new Mail(mail.subject().formatted(args), mail.body().formatted(args));
     }
 
     public static String date(String language, LocalDate date) {
         return (isGerman(language) ? DATE_DE : DATE_EN).format(date);
     }
 
-    private static boolean isGerman(String language) {
+    public static boolean isGerman(String language) {
         return "de".equals(language);
     }
 }

src/main/java/auth_service/modules/mail/model/OutboxRow.java

@@ -1,4 +1,8 @@
 package auth_service.modules.mail.model;
 
-public record OutboxRow(long id, String kind, String toAddress, String subject, String body, int attempts) {
+/**
+ * @param broadcastId null for an account mail
+ */
+public record OutboxRow(long id, String kind, String toAddress, String subject, String body, int attempts,
+                        Long broadcastId) {
 }

src/main/java/auth_service/modules/mail/repo/BroadcastRepository.java

@@ -0,0 +1,83 @@
+package auth_service.modules.mail.repo;
+
+import auth_service.modules.mail.model.Broadcast;
+import auth_service.modules.mail.model.BroadcastText;
+import auth_service.modules.mail.model.Mail;
+import auth_service.shared.account.ActiveUser;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Repository;
+
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+
+@Repository
+public class BroadcastRepository {
+
+    static final String RECIPIENTS = "FROM users u JOIN user_settings s ON s.user_id = u.id WHERE "
+            + ActiveUser.of("u") + " AND s.platform_mails";
+
+    private final JdbcTemplate jdbc;
+    private final NamedParameterJdbcTemplate namedJdbc;
+
+    public BroadcastRepository(JdbcTemplate jdbc, NamedParameterJdbcTemplate namedJdbc) {
+        this.jdbc = jdbc;
+        this.namedJdbc = namedJdbc;
+    }
+
+    public long countRecipients(List<Long> userIds) {
+        return Objects.requireNonNull(namedJdbc.queryForObject("SELECT COUNT(*) " + recipients(userIds),
+                ids(userIds), Long.class));
+    }
+
+    public long insert(BroadcastText text, long createdBy, Instant createdAt) {
+        return Objects.requireNonNull(jdbc.queryForObject("""
+                INSERT INTO mail_broadcast (subject_en, subject_de, body_en, body_de, created_by, created_at)
+                VALUES (?, ?, ?, ?, ?, ?) RETURNING id""", Long.class, text.subjectEn(), text.subjectDe(),
+                text.bodyEn(), text.bodyDe(), createdBy, Timestamp.from(createdAt)));
+    }
+
+    /**
+     * One outbox row per recipient, in the language of the user (English when none is set).
+     *
+     * @return the number of rows queued
+     */
+    public int queue(long broadcastId, List<Long> userIds, Mail en, Mail de) {
+        return namedJdbc.update("""
+                INSERT INTO mail_outbox (kind, user_id, to_address, subject, body, broadcast_id)
+                SELECT 'BROADCAST', u.id, u.email, CASE WHEN s.language = 'de' THEN :subjectDe ELSE :subjectEn END,
+                       CASE WHEN s.language = 'de' THEN :bodyDe ELSE :bodyEn END, :broadcastId
+                """ + recipients(userIds), ids(userIds)
+                .addValue("subjectEn", en.subject()).addValue("bodyEn", en.body())
+                .addValue("subjectDe", de.subject()).addValue("bodyDe", de.body())
+                .addValue("broadcastId", broadcastId));
+    }
+
+    public void setRecipientCount(long broadcastId, int count) {
+        jdbc.update("UPDATE mail_broadcast SET recipient_count = ? WHERE id = ?", count, broadcastId);
+    }
+
+    public List<Broadcast> findAll() {
+        return jdbc.query("SELECT * FROM mail_broadcast ORDER BY created_at DESC, id DESC",
+                (rs, row) -> new Broadcast(rs.getLong("id"),
+                        new BroadcastText(rs.getString("subject_en"), rs.getString("subject_de"),
+                                rs.getString("body_en"), rs.getString("body_de")),
+                        rs.getLong("created_by"), rs.getTimestamp("created_at").toInstant(),
+                        rs.getInt("recipient_count"), rs.getInt("sent_count"), rs.getInt("failed_count")));
+    }
+
+    private static String recipients(List<Long> userIds) {
+        if (userIds == null) {
+            return RECIPIENTS;
+        }
+        // An empty list would make "IN ()", a syntax error.
+        return RECIPIENTS + (userIds.isEmpty() ? " AND FALSE" : " AND u.id IN (:ids)");
+    }
+
+    private static MapSqlParameterSource ids(List<Long> userIds) {
+        return new MapSqlParameterSource("ids", userIds);
+    }
+}

src/main/java/auth_service/modules/mail/repo/MailOutboxRepository.java

@@ -10,6 +10,7 @@ import org.springframework.stereotype.Repository;
 import java.sql.Timestamp;
 import java.time.Instant;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 
 @Repository
@@ -39,16 +40,19 @@ public class MailOutboxRepository {
 
     public List<OutboxRow> findDue(Instant now, int limit) {
         return jdbc.query("""
-                        SELECT id, kind, to_address, subject, body, attempts FROM mail_outbox
+                        SELECT id, kind, to_address, subject, body, attempts, broadcast_id FROM mail_outbox
                         WHERE status = 'PENDING' AND next_attempt_at <= ?
-                        ORDER BY id LIMIT ?""",
+                        ORDER BY broadcast_id NULLS FIRST, id LIMIT ?""",
                 (rs, row) -> new OutboxRow(rs.getLong("id"), rs.getString("kind"), rs.getString("to_address"),
-                        rs.getString("subject"), rs.getString("body"), rs.getInt("attempts")),
+                        rs.getString("subject"), rs.getString("body"), rs.getInt("attempts"), rs.getObject("broadcast_id", Long.class)),
                 Timestamp.from(now), limit);
     }
 
     public void markSent(long id, Instant now) {
-        jdbc.update("UPDATE mail_outbox SET status = 'SENT', sent_at = ?, last_error = NULL WHERE id = ?",
+        jdbc.update("""
+                        WITH row AS (UPDATE mail_outbox SET status = 'SENT', sent_at = ?, last_error = NULL
+                                      WHERE id = ? RETURNING broadcast_id)
+                        UPDATE mail_broadcast b SET sent_count = sent_count + 1 FROM row WHERE b.id = row.broadcast_id""",
                 Timestamp.from(now), id);
     }
 
@@ -58,10 +62,37 @@ public class MailOutboxRepository {
     }
 
     public void markFailed(long id, int attempts, String error) {
-        jdbc.update("UPDATE mail_outbox SET status = 'FAILED', attempts = ?, last_error = ? WHERE id = ?",
+        jdbc.update("""
+                        WITH row AS (UPDATE mail_outbox SET status = 'FAILED', attempts = ?, last_error = ?
+                                      WHERE id = ? RETURNING broadcast_id)
+                        UPDATE mail_broadcast b SET failed_count = failed_count + 1 FROM row WHERE b.id = row.broadcast_id""",
                 attempts, shorten(error), id);
     }
 
+    /**
+     * Fails the pending broadcast rows whose user is no longer active or turned the mails off since the queue time.
+     *
+     * @return the number of rows failed
+     */
+    public int failBroadcastRowsOfInactiveUsers() {
+        return Objects.requireNonNull(jdbc.queryForObject("""
+                WITH skipped AS (
+                    UPDATE mail_outbox o SET status = 'FAILED', last_error = 'recipient inactive or opted out'
+                     WHERE o.status = 'PENDING' AND o.broadcast_id IS NOT NULL
+                       AND NOT EXISTS (SELECT 1 %s AND u.id = o.user_id)
+                    RETURNING o.broadcast_id),
+                counted AS (
+                    UPDATE mail_broadcast b SET failed_count = failed_count + c.n
+                      FROM (SELECT broadcast_id, COUNT(*) AS n FROM skipped GROUP BY broadcast_id) c
+                     WHERE b.id = c.broadcast_id)
+                SELECT COUNT(*) FROM skipped""".formatted(BroadcastRepository.RECIPIENTS), Integer.class));
+    }
+
+    public int countSentSince(Instant since) {
+        return Objects.requireNonNull(jdbc.queryForObject("SELECT COUNT(*) FROM mail_outbox WHERE sent_at >= ?",
+                Integer.class, Timestamp.from(since)));
+    }
+
     public void deleteSentBefore(Instant cutoff) {
         jdbc.update("DELETE FROM mail_outbox WHERE status = 'SENT' AND sent_at < ?", Timestamp.from(cutoff));
     }

src/main/java/auth_service/modules/mail/service/BroadcastService.java

@@ -0,0 +1,82 @@
+package auth_service.modules.mail.service;
+
+import auth_service.modules.mail.exception.NoRecipientsException;
+import auth_service.modules.mail.model.Broadcast;
+import auth_service.modules.mail.model.BroadcastText;
+import auth_service.modules.mail.model.Mail;
+import auth_service.modules.mail.model.MailTemplate;
+import auth_service.modules.mail.model.Recipient;
+import auth_service.modules.mail.repo.BroadcastRepository;
+import auth_service.modules.mail.repo.MailOutboxRepository;
+import auth_service.shared.audit.AdminAuditLog;
+import auth_service.shared.security.RequiresRole;
+import auth_service.shared.security.Role;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.List;
+
+@Service
+@RequiresRole(Role.PLATFORM_ADMIN)
+@Slf4j
+public class BroadcastService {
+
+    private final BroadcastRepository broadcasts;
+    private final MailOutboxRepository outbox;
+    private final AdminAuditLog adminAuditLog;
+    private final String settingsUrl;
+
+    public BroadcastService(BroadcastRepository broadcasts, MailOutboxRepository outbox, AdminAuditLog adminAuditLog,
+                            @Value("${app.frontend-url}") String frontendUrl) {
+        this.broadcasts = broadcasts;
+        this.outbox = outbox;
+        this.adminAuditLog = adminAuditLog;
+        this.settingsUrl = frontendUrl + "/settings";
+    }
+
+    public List<Broadcast> findAll() {
+        return broadcasts.findAll();
+    }
+
+    /**
+     * @param userIds null = every active user with platform mails on
+     */
+    public long countRecipients(List<Long> userIds) {
+        return broadcasts.countRecipients(userIds);
+    }
+
+    /**
+     * @param userIds null = every active user with platform mails on
+     * @return the number of mails queued
+     * @throws NoRecipientsException when nobody is left; nothing is stored then
+     */
+    @Transactional
+    public int send(long actorUserId, BroadcastText text, List<Long> userIds) {
+        Mail en = MailTemplate.BROADCAST.render("en", text.subjectEn(), text.bodyEn(), settingsUrl);
+        Mail de = MailTemplate.BROADCAST.render("de", text.subjectDe(), text.bodyDe(), settingsUrl);
+        long id = broadcasts.insert(text, actorUserId, Instant.now());
+        int queued = broadcasts.queue(id, userIds, en, de);
+        if (queued == 0) {
+            throw new NoRecipientsException();
+        }
+        broadcasts.setRecipientCount(id, queued);
+        adminAuditLog.record(actorUserId, "BROADCAST_SEND", null, String.valueOf(id));
+        log.info("Admin {} queued broadcast {} for {} users", actorUserId, id, queued);
+        return queued;
+    }
+
+    /**
+     * Mails the draft to the admin only, in the admin's language.
+     */
+    @Transactional
+    public void sendTest(long actorUserId, BroadcastText text) {
+        Recipient admin = outbox.findRecipient(actorUserId).orElseThrow();
+        boolean german = MailTemplate.isGerman(admin.language());
+        Mail mail = MailTemplate.BROADCAST.render(admin.language(), german ? text.subjectDe() : text.subjectEn(),
+                german ? text.bodyDe() : text.bodyEn(), settingsUrl);
+        outbox.insert(MailTemplate.BROADCAST, actorUserId, admin.email(), mail);
+    }
+}

src/main/java/auth_service/modules/mail/service/MailOutboxService.java

@@ -26,16 +26,21 @@ public class MailOutboxService {
     static final int MAX_ATTEMPTS = 10;
     static final Duration KEEP_SENT = Duration.ofDays(30);
     private static final int BATCH = 20;
+    // Kept free of broadcast rows, so a registration mail still goes out on a broadcast day.
+    static final int ACCOUNT_RESERVE = 10;
 
     private final MailOutboxRepository repository;
     private final JavaMailSender sender;
     private final String from;
+    private final int dailyLimit;
 
     // An empty SPRING_MAIL_HOST still creates Boot's sender bean, so the host decides, not the bean.
     public MailOutboxService(MailOutboxRepository repository, ObjectProvider<JavaMailSender> sender,
                              @Value("${spring.mail.host:}") String host, @Value("${app.mail.from}") String from,
-                             @Value("${app.mail.admin-to}") String adminTo) {
+                             @Value("${app.mail.admin-to}") String adminTo,
+                             @Value("${app.mail.daily-limit}") int dailyLimit) {
         this.repository = repository;
+        this.dailyLimit = dailyLimit;
         this.from = from;
         if (host.isBlank()) {
             this.sender = null;
@@ -45,6 +50,10 @@ public class MailOutboxService {
         if (from.isBlank() || adminTo.isBlank()) {
             throw new IllegalStateException("SPRING_MAIL_HOST is set, so APP_MAIL_FROM and APP_MAIL_ADMIN_TO are needed");
         }
+        if (dailyLimit <= ACCOUNT_RESERVE) {
+            throw new IllegalStateException("APP_MAIL_DAILY_LIMIT must be above " + ACCOUNT_RESERVE
+                    + ", the room kept for account mails");
+        }
         this.sender = sender.getObject();
     }
 
@@ -57,8 +66,18 @@ public class MailOutboxService {
             return;
         }
         Instant now = Instant.now();
-        for (OutboxRow row : repository.findDue(now, BATCH)) {
+        int skipped = repository.failBroadcastRowsOfInactiveUsers();
+        if (skipped > 0) {
+            log.info("{} broadcast mails skipped: recipient inactive or opted out", skipped);
+        }
+        int budget = dailyLimit - repository.countSentSince(now.minus(Duration.ofHours(24)));
+        for (OutboxRow row : repository.findDue(now, Math.min(BATCH, Math.max(budget, 0)))) {
+            // Account mails come first, so from the first broadcast row on only broadcast rows follow.
+            if (row.broadcastId() != null && budget <= ACCOUNT_RESERVE) {
+                break;
+            }
             send(row, now);
+            budget--; // a failed try counts too: safer than Hostinger's own count
         }
         repository.deleteSentBefore(now.minus(KEEP_SENT));
     }

src/main/java/auth_service/modules/user/api/UserController.java

@@ -4,6 +4,7 @@ import auth_service.shared.security.UserPrincipal;
 import auth_service.modules.user.api.dto.request.AppearanceRequest;
 import auth_service.modules.user.api.dto.request.DashboardTilesRequest;
 import auth_service.modules.user.api.dto.request.FavoritePlatformRequest;
+import auth_service.modules.user.api.dto.request.PlatformMailsRequest;
 import auth_service.modules.user.api.dto.request.ProfileTilesRequest;
 import auth_service.modules.user.api.dto.request.ProfileStyleRequest;
 import auth_service.modules.user.api.dto.request.UserSettingsRequest;
@@ -64,6 +65,13 @@ public class UserController {
         return ResponseEntity.noContent().build();
     }
 
+    @PutMapping(value = "/settings/platform-mails")
+    public ResponseEntity<Void> updatePlatformMails(@AuthenticationPrincipal UserPrincipal principal,
+                                                    @Valid @RequestBody PlatformMailsRequest request) {
+        this.userService.updatePlatformMails(principal.getIdAsLong(), request.enabled());
+        return ResponseEntity.noContent().build();
+    }
+
     @PutMapping(value = "/settings/welcome-seen")
     public ResponseEntity<Void> markWelcomeSeen(@AuthenticationPrincipal UserPrincipal principal) {
         this.userService.markWelcomeSeen(principal.getIdAsLong());

src/main/java/auth_service/modules/user/api/dto/request/PlatformMailsRequest.java

@@ -0,0 +1,9 @@
+package auth_service.modules.user.api.dto.request;
+
+import jakarta.validation.constraints.NotNull;
+
+/**
+ * @param enabled boxed, so a missing field is a 400 and not an opt-out
+ */
+public record PlatformMailsRequest(@NotNull Boolean enabled) {
+}

src/main/java/auth_service/modules/user/api/dto/response/SettingsResponse.java

@@ -9,5 +9,5 @@ import java.util.List;
 public record SettingsResponse(long id, boolean useDarkTheme, String language, boolean publicProfile,
                                AppearanceResponse appearance, ProfileStyleResponse profileStyle,
                                List<TileResponse> dashboardTiles, List<TileResponse> profileTiles,
-                               String favoritePlatform, Instant welcomeSeenAt) {
+                               String favoritePlatform, boolean platformMails, Instant welcomeSeenAt) {
 }

src/main/java/auth_service/modules/user/api/mapper/ApiMapper.java

@@ -48,6 +48,7 @@ public abstract class ApiMapper {
                                 : toApi(user.getSettings().getDashboardTiles()))
                         .profileTiles(toApi(user.getSettings().getProfileTiles()))
                         .favoritePlatform(user.getSettings().getFavoritePlatform())
+                        .platformMails(user.getSettings().isPlatformMails())
                         .welcomeSeenAt(user.getSettings().getWelcomeSeenAt())
                         .build())
                 .build();

src/main/java/auth_service/modules/user/model/UserSettings.java

@@ -21,6 +21,9 @@ public class UserSettings {
     @Builder.Default
     private List<Tile> profileTiles = DEFAULT_PROFILE_TILES;
     private String favoritePlatform;
+    /** false = no broadcast mails; account mails always go. */
+    @Builder.Default
+    private boolean platformMails = true;
     /** Null: the welcome page was not seen yet. */
     private Instant welcomeSeenAt;
 

src/main/java/auth_service/modules/user/persistence/entities/UserSettingsEntity.java

@@ -70,6 +70,9 @@ public class UserSettingsEntity {
     @Column(name = "favorite_platform")
     private String favoritePlatform;
 
+    @Column(name = "platform_mails", nullable = false)
+    private boolean platformMails;
+
     @JdbcTypeCode(SqlTypes.ARRAY)
     @Column(name = "profile_tiles", nullable = false)
     private List<String> profileTiles;

src/main/java/auth_service/modules/user/persistence/mapper/EntityMapper.java

@@ -35,6 +35,7 @@ public abstract class EntityMapper {
                         : settings.getDashboardTiles().stream().map(Tile::parse).toList())
                 .profileTiles(settings.getProfileTiles().stream().map(Tile::parse).toList())
                 .favoritePlatform(settings.getFavoritePlatform())
+                .platformMails(settings.isPlatformMails())
                 .welcomeSeenAt(settings.getWelcomeSeenAt())
                 .build();
 
@@ -74,6 +75,7 @@ public abstract class EntityMapper {
                 : Tile.storedAll(settings.getDashboardTiles()));
         target.setProfileTiles(Tile.storedAll(settings.getProfileTiles()));
         target.setFavoritePlatform(settings.getFavoritePlatform());
+        target.setPlatformMails(settings.isPlatformMails());
         target.setWelcomeSeenAt(settings.getWelcomeSeenAt());
     }
 }

src/main/java/auth_service/modules/user/service/UserService.java

@@ -129,6 +129,13 @@ public class UserService {
                 user -> user.getSettings().setFavoritePlatform(platform == null ? null : platform.strip()));
     }
 
+    @RequiresRole(Role.USER)
+    @Transactional
+    @CacheEvict(value = "users", key = "#userId")
+    public void updatePlatformMails(long userId, boolean enabled) {
+        userRepository.update(userId, user -> user.getSettings().setPlatformMails(enabled));
+    }
+
     @RequiresRole(Role.USER)
     @Transactional
     @CacheEvict(value = "users", key = "#userId")

src/main/resources/application.yml

@@ -83,3 +83,6 @@ app:
     # Gets "a new registration waits". Only in the env file, never in git.
     admin-to: ${APP_MAIL_ADMIN_TO:}
     cron: "0 * * * * *"
+    # Mails per rolling 24 h, account mails and broadcasts together. Hostinger: 100 for the free mailbox of a
+    # hosting plan, 1000 Business Starter, 3000 Business Standard / Premium. Broadcast rows keep 10 of them free.
+    daily-limit: ${APP_MAIL_DAILY_LIMIT:100}

src/main/resources/db/migration/V25__Broadcast_Mail.sql

@@ -0,0 +1,21 @@
+-- "Mails from the platform" in the settings. Off = no broadcast; account mails always go.
+ALTER TABLE user_settings ADD COLUMN platform_mails BOOLEAN NOT NULL DEFAULT TRUE;
+
+-- One row per broadcast an admin sent: the texts and the counts. No FK on created_by (the row outlives the admin).
+CREATE TABLE mail_broadcast (
+    id              BIGSERIAL    PRIMARY KEY,
+    subject_en      VARCHAR(150) NOT NULL,
+    subject_de      VARCHAR(150) NOT NULL,
+    body_en         TEXT         NOT NULL,
+    body_de         TEXT         NOT NULL,
+    created_by      BIGINT       NOT NULL,
+    created_at      TIMESTAMPTZ  NOT NULL,
+    recipient_count INT          NOT NULL DEFAULT 0,
+    sent_count      INT          NOT NULL DEFAULT 0,
+    failed_count    INT          NOT NULL DEFAULT 0
+);
+
+-- A broadcast row of the outbox; NULL = account mail. The sender takes account mails first.
+ALTER TABLE mail_outbox ADD COLUMN broadcast_id BIGINT REFERENCES mail_broadcast (id) ON DELETE CASCADE;
+-- One mail per user and broadcast, whatever the retries do.
+CREATE UNIQUE INDEX uq_mail_outbox_broadcast_user ON mail_outbox (broadcast_id, user_id) WHERE broadcast_id IS NOT NULL;

src/test/java/auth_service/modules/mail/api/BroadcastFlowTest.java

@@ -0,0 +1,172 @@
+package auth_service.modules.mail.api;
+
+import auth_service.configuration.security.ClientHeaderFilter;
+import auth_service.modules.mail.service.MailOutboxService;
+import auth_service.shared.api.AuthCookies;
+import auth_service.shared.test.AbstractIntegrationTest;
+import jakarta.servlet.http.Cookie;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.MediaType;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.mail.SimpleMailMessage;
+import org.springframework.security.crypto.password.PasswordEncoder;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.ResultActions;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+class BroadcastFlowTest extends AbstractIntegrationTest {
+
+    private static final String PASSWORD = "correct horse battery";
+    private static final String ADMIN = "broadcast-flow-admin@example.com";
+    private static final String PLAYER = "broadcast-flow-player@example.com";
+    private static final String OPTED_OUT = "broadcast-flow-optedout@example.com";
+    private static final String BLOCKED = "broadcast-flow-blocked@example.com";
+    private static final String BASE = "/api/admin/broadcasts";
+
+    @Autowired
+    private MockMvc mockMvc;
+
+    @Autowired
+    private JdbcTemplate jdbc;
+
+    @Autowired
+    private PasswordEncoder passwordEncoder;
+
+    @Autowired
+    private MailOutboxService mailOutboxService;
+
+    private long optedOutId;
+    private List<Long> foreignOptedIn;
+
+    private long insertUser(String email, String role, String language, boolean platformMails, boolean enabled) {
+        long id = jdbc.queryForObject("INSERT INTO users (email, password, enabled, role, display_name) "
+                        + "VALUES (?, ?, ?, ?, ?) RETURNING id", Long.class, email, passwordEncoder.encode(PASSWORD),
+                enabled, role, email.substring(15, 20));
+        addSettingsRows();
+        jdbc.update("UPDATE user_settings SET language = ?, platform_mails = ? WHERE user_id = ?", language,
+                platformMails, id);
+        return id;
+    }
+
+    @BeforeEach
+    void seed() {
+        jdbc.update("DELETE FROM mail_outbox");
+        jdbc.update("DELETE FROM mail_broadcast");
+        jdbc.update("DELETE FROM users WHERE email LIKE 'broadcast-flow-%'");
+        // Users that other test classes left behind must not count as recipients; tearDown restores them.
+        foreignOptedIn = jdbc.queryForList("SELECT user_id FROM user_settings WHERE platform_mails", Long.class);
+        jdbc.update("UPDATE user_settings SET platform_mails = FALSE");
+        insertUser(ADMIN, "PLATFORM_ADMIN", "de", true, true);
+        insertUser(PLAYER, "USER", "en", true, true);
+        optedOutId = insertUser(OPTED_OUT, "USER", "en", false, true);
+        insertUser(BLOCKED, "USER", "en", true, true);
+        jdbc.update("UPDATE user_settings SET platform_mails = TRUE WHERE user_id IN "
+                + "(SELECT id FROM users WHERE email LIKE 'broadcast-flow-%' AND email <> ?)", OPTED_OUT);
+        jdbc.update("UPDATE users SET enabled = FALSE WHERE email = ?", BLOCKED);
+    }
+
+    @AfterEach
+    void tearDown() {
+        jdbc.update("DELETE FROM users WHERE email LIKE 'broadcast-flow-%'");
+        jdbc.batchUpdate("UPDATE user_settings SET platform_mails = TRUE WHERE user_id = ?",
+                foreignOptedIn.stream().map(id -> new Object[]{id}).toList());
+    }
+
+    private Cookie login(String email) throws Exception {
+        return new Cookie(AuthCookies.ACCESS_COOKIE, mockMvc.perform(post("/api/auth/login")
+                        .header(ClientHeaderFilter.CLIENT_HEADER, "web").contentType(MediaType.APPLICATION_JSON)
+                        .content("{\"email\":\"" + email + "\",\"password\":\"" + PASSWORD + "\"}"))
+                .andReturn().getResponse().getCookie(AuthCookies.ACCESS_COOKIE).getValue());
+    }
+
+    private static String broadcast(String subjectDe, String userIds) {
+        return "{\"subjectEn\":\"Hello\",\"subjectDe\":\"" + subjectDe + "\",\"bodyEn\":\"Body\",\"bodyDe\":\"Inhalt\""
+                + ",\"userIds\":" + userIds + "}";
+    }
+
+    private ResultActions send(MockHttpServletRequestBuilder request, Cookie session) throws Exception {
+        return mockMvc.perform(request.header(ClientHeaderFilter.CLIENT_HEADER, "web").cookie(session)
+                .contentType(MediaType.APPLICATION_JSON));
+    }
+
+    private List<Map<String, Object>> rows() {
+        return jdbc.queryForList("SELECT to_address, subject, kind, broadcast_id FROM mail_outbox ORDER BY to_address");
+    }
+
+    @Test
+    void sendToAll_MailsActiveOptedInUsersInTheirLanguageAndCountsTheSend() throws Exception {
+        Cookie admin = login(ADMIN);
+
+        send(get(BASE + "/recipient-count"), admin).andExpect(status().isOk()).andExpect(jsonPath("$.count").value(2));
+        send(post(BASE).content(broadcast("Hallo", "null")), admin).andExpect(status().isCreated())
+                .andExpect(jsonPath("$.recipientCount").value(2));
+
+        assertThat(rows()).extracting(row -> row.get("to_address") + " " + row.get("subject") + " " + row.get("kind"))
+                .containsExactly(ADMIN + " Hallo BROADCAST", PLAYER + " Hello BROADCAST");
+
+        jdbc.update("UPDATE users SET enabled = FALSE WHERE email = ?", PLAYER);
+        mailOutboxService.sendDue();
+
+        verify(mailSender, times(1)).send(any(SimpleMailMessage.class));
+        send(get(BASE), admin).andExpect(status().isOk())
+                .andExpect(jsonPath("$.results[0].subjectDe").value("Hallo"))
+                .andExpect(jsonPath("$.results[0].recipientCount").value(2))
+                .andExpect(jsonPath("$.results[0].sentCount").value(1))
+                .andExpect(jsonPath("$.results[0].failedCount").value(1));
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM admin_audit WHERE action = 'BROADCAST_SEND'", Long.class))
+                .isEqualTo(1);
+    }
+
+    @Test
+    void pickedAndTest_StayNarrow() throws Exception {
+        Cookie admin = login(ADMIN);
+
+        send(post(BASE).content(broadcast("Hallo", "[" + optedOutId + "]")), admin).andExpect(status().isConflict())
+                .andExpect(jsonPath("$.code").value("NO_RECIPIENTS"));
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM mail_broadcast", Long.class)).isZero();
+
+        send(post(BASE + "/test").content(broadcast("Hallo", "null")), admin).andExpect(status().isNoContent());
+        assertThat(rows()).extracting(row -> row.get("to_address") + " " + row.get("subject") + " " + row.get("broadcast_id"))
+                .containsExactly(ADMIN + " Hallo null");
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM mail_broadcast", Long.class)).isZero();
+
+        send(post(BASE).content(broadcast(" ", "null")), admin).andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code").value("VALIDATION_FAILED"));
+    }
+
+    @Test
+    void everyEndpoint_RefusesUsersAndStrangers() throws Exception {
+        Cookie player = login(PLAYER);
+        String body = broadcast("Hallo", "null");
+
+        send(get(BASE), player).andExpect(status().isForbidden());
+        send(get(BASE + "/recipient-count"), player).andExpect(status().isForbidden());
+        send(post(BASE).content(body), player).andExpect(status().isForbidden());
+        send(post(BASE + "/test").content(body), player).andExpect(status().isForbidden());
+
+        mockMvc.perform(get(BASE).header(ClientHeaderFilter.CLIENT_HEADER, "web")).andExpect(status().isUnauthorized());
+        mockMvc.perform(get(BASE + "/recipient-count").header(ClientHeaderFilter.CLIENT_HEADER, "web")).andExpect(status().isUnauthorized());
+        mockMvc.perform(post(BASE).header(ClientHeaderFilter.CLIENT_HEADER, "web").contentType(MediaType.APPLICATION_JSON).content(body))
+                .andExpect(status().isUnauthorized());
+        mockMvc.perform(post(BASE + "/test").header(ClientHeaderFilter.CLIENT_HEADER, "web").contentType(MediaType.APPLICATION_JSON).content(body))
+                .andExpect(status().isUnauthorized());
+
+        assertThat(rows()).isEmpty();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM mail_broadcast", Long.class)).isZero();
+    }
+}

src/test/java/auth_service/modules/mail/model/MailTemplateTest.java

@@ -19,8 +19,20 @@ class MailTemplateTest extends AbstractUnitTest {
         for (String language : new String[]{"de", "en"}) {
             Mail mail = template.render(language, "Alice", "alice@example.com", "https://tbp.example");
 
-            assertThat(mail.subject()).isNotBlank().doesNotContain("%");
-            assertThat(mail.body()).contains("Alice").doesNotContain("%s");
+            // BROADCAST puts its first argument into the subject, so look at both.
+            String text = mail.subject() + mail.body();
+            assertThat(mail.subject()).isNotBlank();
+            assertThat(text).contains("Alice").doesNotContain("%");
+        }
+    }
+
+    @Test
+    void render_Broadcast_PutsSubjectBodyAndTheSettingsLink() {
+        for (String language : new String[]{"de", "en"}) {
+            Mail mail = MailTemplate.BROADCAST.render(language, "50% off", "Hello 100%s", "https://tbp.example/settings");
+
+            assertThat(mail.subject()).isEqualTo("50% off");
+            assertThat(mail.body()).startsWith("Hello 100%s").endsWith("https://tbp.example/settings\n");
         }
     }
 

src/test/java/auth_service/modules/mail/repo/BroadcastRepositoryTest.java

@@ -0,0 +1,101 @@
+package auth_service.modules.mail.repo;
+
+import auth_service.modules.mail.model.Broadcast;
+import auth_service.modules.mail.model.BroadcastText;
+import auth_service.modules.mail.model.Mail;
+import auth_service.shared.test.AbstractPersistenceTest;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BroadcastRepositoryTest extends AbstractPersistenceTest {
+
+    @Autowired
+    private BroadcastRepository repository;
+
+    @Autowired
+    private JdbcTemplate jdbc;
+
+    private long german;
+    private long noLanguage;
+    private long optedOut;
+    private long blocked;
+    private long pending;
+    private long deleted;
+
+    @BeforeEach
+    void seed() {
+        jdbc.update("DELETE FROM mail_outbox");
+        jdbc.update("DELETE FROM mail_broadcast");
+        jdbc.update("DELETE FROM users WHERE email LIKE 'bc-repo-%'");
+        german = user("german", "true, 'ACTIVE', NULL", "de", true);
+        noLanguage = user("nolang", "true, 'ACTIVE', NULL", "", true);
+        optedOut = user("optedout", "true, 'ACTIVE', NULL", "de", false);
+        blocked = user("blocked", "false, 'ACTIVE', NULL", "de", true);
+        pending = user("pending", "true, 'PENDING', NULL", "de", true);
+        deleted = user("deleted", "true, 'ACTIVE', NOW()", "de", true);
+    }
+
+    private long user(String name, String flags, String language, boolean platformMails) {
+        long id = jdbc.queryForObject("INSERT INTO users (email, password, enabled, status, deleted_at) VALUES (?, 'x', "
+                + flags + ") RETURNING id", Long.class, "bc-repo-" + name + "@example.com");
+        jdbc.update("INSERT INTO user_settings (user_id, language, platform_mails) VALUES (?, ?, ?)", id, language,
+                platformMails);
+        return id;
+    }
+
+    @Test
+    void countRecipients_TakesOnlyActiveOptedInUsers() {
+        // Other tests of the suite may leave users behind, so compare with the same query by SQL.
+        long all = jdbc.queryForObject("SELECT COUNT(*) FROM users u JOIN user_settings s ON s.user_id = u.id "
+                + "WHERE u.enabled AND u.deleted_at IS NULL AND u.status = 'ACTIVE' AND s.platform_mails", Long.class);
+
+        assertThat(repository.countRecipients(null)).isEqualTo(all).isGreaterThanOrEqualTo(2);
+        assertThat(repository.countRecipients(List.of(german, blocked, optedOut, pending, deleted))).isEqualTo(1);
+        assertThat(repository.countRecipients(List.of())).isZero();
+    }
+
+    @Test
+    void queue_WritesOneRowPerRecipientInTheirLanguage() {
+        long id = repository.insert(new BroadcastText("Hi", "Hallo", "Body", "Inhalt"), 1L, Instant.now());
+
+        int queued = repository.queue(id, List.of(german, noLanguage, optedOut, blocked),
+                new Mail("Hi", "Body"), new Mail("Hallo", "Inhalt"));
+        repository.setRecipientCount(id, queued);
+
+        assertThat(queued).isEqualTo(2);
+        List<Map<String, Object>> rows = jdbc.queryForList("SELECT user_id, kind, subject, body, broadcast_id "
+                + "FROM mail_outbox ORDER BY user_id");
+        assertThat(rows).extracting(row -> row.get("user_id")).containsExactly(german, noLanguage);
+        assertThat(rows).extracting(row -> row.get("subject")).containsExactly("Hallo", "Hi");
+        assertThat(rows).extracting(row -> row.get("body")).containsExactly("Inhalt", "Body");
+        assertThat(rows).extracting(row -> row.get("kind")).containsOnly("BROADCAST");
+        assertThat(rows).extracting(row -> row.get("broadcast_id")).containsOnly(id);
+
+        long newer = repository.insert(new BroadcastText("a", "b", "c", "d"), 1L, Instant.now().plusSeconds(5));
+        List<Broadcast> all = repository.findAll();
+        assertThat(all).extracting(Broadcast::id).containsExactly(newer, id);
+        assertThat(all.get(1)).satisfies(b -> {
+            assertThat(b.recipientCount()).isEqualTo(2);
+            assertThat(b.sentCount()).isZero();
+            assertThat(b.failedCount()).isZero();
+            assertThat(b.text().subjectDe()).isEqualTo("Hallo");
+        });
+    }
+
+    @Test
+    void queue_NeverWritesTwoRowsForOneUserAndBroadcast() {
+        long id = repository.insert(new BroadcastText("a", "b", "c", "d"), 1L, Instant.now());
+        repository.queue(id, List.of(german), new Mail("a", "c"), new Mail("b", "d"));
+
+        org.junit.jupiter.api.Assertions.assertThrows(org.springframework.dao.DataIntegrityViolationException.class,
+                () -> repository.queue(id, List.of(german), new Mail("a", "c"), new Mail("b", "d")));
+    }
+}

src/test/java/auth_service/modules/mail/repo/MailOutboxRepositoryTest.java

@@ -30,6 +30,7 @@ class MailOutboxRepositoryTest extends AbstractPersistenceTest {
     @BeforeEach
     void clean() {
         jdbc.update("DELETE FROM mail_outbox");
+        jdbc.update("DELETE FROM mail_broadcast");
         jdbc.update("DELETE FROM users WHERE email LIKE 'mail-repo%'");
         userId = jdbc.queryForObject("INSERT INTO users (email, password, enabled) "
                 + "VALUES ('mail-repo-user@example.com', 'x', true) RETURNING id", Long.class);
@@ -49,7 +50,7 @@ class MailOutboxRepositoryTest extends AbstractPersistenceTest {
         long id = insert();
 
         assertThat(repository.findDue(now.plusSeconds(1), 10))
-                .containsExactly(new OutboxRow(id, "ACCOUNT_BLOCKED", "alice@example.com", "Subject", "Body", 0));
+                .containsExactly(new OutboxRow(id, "ACCOUNT_BLOCKED", "alice@example.com", "Subject", "Body", 0, null));
     }
 
     @Test
@@ -117,6 +118,85 @@ class MailOutboxRepositoryTest extends AbstractPersistenceTest {
         assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM mail_outbox", Integer.class)).isZero();
     }
 
+    private long broadcast() {
+        return jdbc.queryForObject("INSERT INTO mail_broadcast (subject_en, subject_de, body_en, body_de, created_by, "
+                + "created_at) VALUES ('s', 's', 'b', 'b', 1, NOW()) RETURNING id", Long.class);
+    }
+
+    private long broadcastRow(long broadcastId, long forUser) {
+        return jdbc.queryForObject("INSERT INTO mail_outbox (kind, user_id, to_address, subject, body, broadcast_id) "
+                + "VALUES ('BROADCAST', ?, 'x@example.com', 's', 'b', ?) RETURNING id", Long.class, forUser, broadcastId);
+    }
+
+    private long user(String name, String flags) {
+        return jdbc.queryForObject("INSERT INTO users (email, password, enabled, status, deleted_at) "
+                + "VALUES (?, 'x', " + flags + ") RETURNING id", Long.class, "mail-repo-" + name + "@example.com");
+    }
+
+    private int counter(long broadcastId, String column) {
+        return jdbc.queryForObject("SELECT " + column + " FROM mail_broadcast WHERE id = ?", Integer.class, broadcastId);
+    }
+
+    @Test
+    void findDue_PutsAccountMailsBeforeBroadcastRows() {
+        long broadcastRow = broadcastRow(broadcast(), userId);
+        long account = insert();
+
+        assertThat(repository.findDue(now.plusSeconds(1), 10)).extracting(OutboxRow::id)
+                .containsExactly(account, broadcastRow);
+    }
+
+    @Test
+    void markSentAndMarkFailed_CountOnTheBroadcast() {
+        long broadcast = broadcast();
+        long sent = broadcastRow(broadcast, userId);
+        long failed = broadcastRow(broadcast, user("other", "true, 'ACTIVE', NULL"));
+        long account = insert();
+
+        repository.markSent(sent, now);
+        repository.markFailed(failed, 10, "rejected");
+        repository.markSent(account, now);
+
+        assertThat(counter(broadcast, "sent_count")).isEqualTo(1);
+        assertThat(counter(broadcast, "failed_count")).isEqualTo(1);
+    }
+
+    @Test
+    void failBroadcastRowsOfInactiveUsers_FailsBlockedDeletedPendingAndOptedOut() {
+        long broadcast = broadcast();
+        long active = user("active", "true, 'ACTIVE', NULL");
+        long blocked = user("blocked", "false, 'ACTIVE', NULL");
+        long deleted = user("deleted", "true, 'ACTIVE', NOW()");
+        long pending = user("pending", "true, 'PENDING', NULL");
+        long optedOut = user("optedout", "true, 'ACTIVE', NULL");
+        for (long id : new long[]{active, blocked, deleted, pending, optedOut}) {
+            jdbc.update("INSERT INTO user_settings (user_id, platform_mails) VALUES (?, ?)", id, id != optedOut);
+        }
+        long keep = broadcastRow(broadcast, active);
+        for (long id : new long[]{blocked, deleted, pending, optedOut}) {
+            broadcastRow(broadcast, id);
+        }
+        long accountOfBlocked = jdbc.queryForObject("INSERT INTO mail_outbox (kind, user_id, to_address, subject, body) "
+                + "VALUES ('ACCOUNT_BLOCKED', ?, 'x@example.com', 's', 'b') RETURNING id", Long.class, blocked);
+
+        assertThat(repository.failBroadcastRowsOfInactiveUsers()).isEqualTo(4);
+
+        assertThat(jdbc.queryForList("SELECT id FROM mail_outbox WHERE status = 'PENDING' ORDER BY id", Long.class))
+                .containsExactly(keep, accountOfBlocked);
+        assertThat(counter(broadcast, "failed_count")).isEqualTo(4);
+    }
+
+    @Test
+    void countSentSince_CountsOnlySentRowsAfterTheInstant() {
+        long recent = insert();
+        long old = insert();
+        insert();
+        repository.markSent(recent, now);
+        repository.markSent(old, now.minus(Duration.ofHours(25)));
+
+        assertThat(repository.countSentSince(now.minus(Duration.ofHours(24)))).isEqualTo(1);
+    }
+
     @Test
     void findRecipient_ReadsEmailNameAndLanguage() {
         long id = jdbc.queryForObject("INSERT INTO users (email, password, enabled, display_name) "

src/test/java/auth_service/modules/mail/service/BroadcastServiceTest.java

@@ -0,0 +1,95 @@
+package auth_service.modules.mail.service;
+
+import auth_service.modules.mail.exception.NoRecipientsException;
+import auth_service.modules.mail.model.BroadcastText;
+import auth_service.modules.mail.model.Mail;
+import auth_service.modules.mail.model.MailTemplate;
+import auth_service.modules.mail.model.Recipient;
+import auth_service.modules.mail.repo.BroadcastRepository;
+import auth_service.modules.mail.repo.MailOutboxRepository;
+import auth_service.shared.audit.AdminAuditLog;
+import auth_service.shared.test.AbstractUnitTest;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class BroadcastServiceTest extends AbstractUnitTest {
+
+    private static final long ADMIN = 1L;
+    private static final String SETTINGS = "https://tbp.example/settings";
+    private static final BroadcastText TEXT = new BroadcastText("Hi", "Hallo", "Body", "Inhalt");
+
+    @Mock
+    private BroadcastRepository broadcasts;
+
+    @Mock
+    private MailOutboxRepository outbox;
+
+    @Mock
+    private AdminAuditLog adminAuditLog;
+
+    private BroadcastService service;
+
+    @BeforeEach
+    void setUp() {
+        service = new BroadcastService(broadcasts, outbox, adminAuditLog, "https://tbp.example");
+    }
+
+    @Test
+    void send_QueuesBothLanguagesAuditsAndReturnsTheCount() {
+        when(broadcasts.insert(any(), anyLong(), any())).thenReturn(5L);
+        when(broadcasts.queue(eq(5L), isNull(), any(), any())).thenReturn(3);
+
+        assertThat(service.send(ADMIN, TEXT, null)).isEqualTo(3);
+
+        ArgumentCaptor<Mail> en = ArgumentCaptor.forClass(Mail.class);
+        ArgumentCaptor<Mail> de = ArgumentCaptor.forClass(Mail.class);
+        verify(broadcasts).queue(eq(5L), isNull(), en.capture(), de.capture());
+        assertThat(en.getValue().subject()).isEqualTo("Hi");
+        assertThat(en.getValue().body()).startsWith("Body").contains("You get this mail").endsWith(SETTINGS + "\n");
+        assertThat(de.getValue().subject()).isEqualTo("Hallo");
+        assertThat(de.getValue().body()).startsWith("Inhalt").contains("Du bekommst diese Mail").endsWith(SETTINGS + "\n");
+        verify(broadcasts).setRecipientCount(5L, 3);
+        verify(adminAuditLog).record(ADMIN, "BROADCAST_SEND", null, "5");
+    }
+
+    @Test
+    void send_RefusesZeroRecipients() {
+        when(broadcasts.insert(any(), anyLong(), any())).thenReturn(5L);
+        when(broadcasts.queue(anyLong(), any(), any(), any())).thenReturn(0);
+
+        assertThatThrownBy(() -> service.send(ADMIN, TEXT, List.of(9L))).isInstanceOf(NoRecipientsException.class);
+
+        verify(broadcasts, never()).setRecipientCount(anyLong(), anyInt());
+        verifyNoInteractions(adminAuditLog);
+    }
+
+    @Test
+    void sendTest_MailsOnlyTheAdminInTheirLanguage() {
+        when(outbox.findRecipient(ADMIN)).thenReturn(Optional.of(new Recipient("admin@example.com", "Admin", "de")));
+
+        service.sendTest(ADMIN, TEXT);
+
+        ArgumentCaptor<Mail> mail = ArgumentCaptor.forClass(Mail.class);
+        verify(outbox).insert(eq(MailTemplate.BROADCAST), eq(ADMIN), eq("admin@example.com"), mail.capture());
+        assertThat(mail.getValue().subject()).isEqualTo("Hallo");
+        assertThat(mail.getValue().body()).startsWith("Inhalt");
+        verifyNoInteractions(broadcasts, adminAuditLog);
+    }
+}

src/test/java/auth_service/modules/mail/service/MailOutboxServiceTest.java

@@ -5,6 +5,7 @@ import auth_service.modules.mail.repo.MailOutboxRepository;
 import auth_service.shared.test.AbstractUnitTest;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.mail.MailSendException;
@@ -22,6 +23,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
@@ -43,11 +45,15 @@ class MailOutboxServiceTest extends AbstractUnitTest {
 
     private MailOutboxService service() {
         when(senderProvider.getObject()).thenReturn(sender);
-        return new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, ADMIN);
+        return new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, ADMIN, 100);
     }
 
     private static OutboxRow row(long id, int attempts) {
-        return new OutboxRow(id, "ACCOUNT_BLOCKED", "alice@example.com", "Subject", "Body", attempts);
+        return new OutboxRow(id, "ACCOUNT_BLOCKED", "alice@example.com", "Subject", "Body", attempts, null);
+    }
+
+    private static OutboxRow broadcastRow(long id) {
+        return new OutboxRow(id, "BROADCAST", "alice@example.com", "Subject", "Body", 0, 1L);
     }
 
     @Test
@@ -125,7 +131,7 @@ class MailOutboxServiceTest extends AbstractUnitTest {
 
     @Test
     void sendDue_DoesNothingWithoutAHost() {
-        MailOutboxService off = new MailOutboxService(repository, senderProvider, "", "", "");
+        MailOutboxService off = new MailOutboxService(repository, senderProvider, "", "", "", 100);
 
         off.sendDue();
 
@@ -134,9 +140,54 @@ class MailOutboxServiceTest extends AbstractUnitTest {
 
     @Test
     void constructor_RefusesAHostWithoutSenderOrAdminAddress() {
-        assertThatThrownBy(() -> new MailOutboxService(repository, senderProvider, "smtp.example.com", "", ADMIN))
+        assertThatThrownBy(() -> new MailOutboxService(repository, senderProvider, "smtp.example.com", "", ADMIN, 100))
                 .isInstanceOf(IllegalStateException.class).hasMessageContaining("APP_MAIL_FROM");
-        assertThatThrownBy(() -> new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, " "))
+        assertThatThrownBy(() -> new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, " ", 100))
                 .isInstanceOf(IllegalStateException.class).hasMessageContaining("APP_MAIL_ADMIN_TO");
     }
+
+    @Test
+    void sendDue_SendsAccountMailsAndStopsBroadcastRowsAtTheReserve() {
+        when(senderProvider.getObject()).thenReturn(sender);
+        MailOutboxService small = new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, ADMIN, 12);
+        when(repository.countSentSince(any())).thenReturn(0);
+        when(repository.findDue(any(), anyInt()))
+                .thenReturn(List.of(row(1, 0), broadcastRow(2), broadcastRow(3), broadcastRow(4)));
+
+        small.sendDue();
+
+        verify(repository).markSent(eq(1L), any());
+        verify(repository).markSent(eq(2L), any());
+        verify(repository, never()).markSent(eq(3L), any());
+        verify(repository, never()).markSent(eq(4L), any());
+    }
+
+    @Test
+    void sendDue_SendsNothingAboveTheDailyLimit() {
+        when(repository.countSentSince(any())).thenReturn(100);
+        when(repository.findDue(any(), eq(0))).thenReturn(List.of());
+
+        service().sendDue();
+
+        verify(repository).findDue(any(), eq(0));
+        verifyNoInteractions(sender);
+    }
+
+    @Test
+    void sendDue_FailsRowsOfInactiveUsersBeforeTheSend() {
+        when(repository.findDue(any(), anyInt())).thenReturn(List.of());
+
+        service().sendDue();
+
+        InOrder order = inOrder(repository);
+        order.verify(repository).failBroadcastRowsOfInactiveUsers();
+        order.verify(repository).findDue(any(), anyInt());
+    }
+
+    @Test
+    void constructorRefusesADailyLimitAtOrBelowTheReserve() {
+        assertThatThrownBy(() -> new MailOutboxService(repository, senderProvider, "smtp.example.com", FROM, ADMIN,
+                MailOutboxService.ACCOUNT_RESERVE))
+                .isInstanceOf(IllegalStateException.class).hasMessageContaining("APP_MAIL_DAILY_LIMIT");
+    }
 }

src/test/java/auth_service/modules/user/api/UserControllerTest.java

@@ -490,6 +490,22 @@ class UserControllerTest extends AbstractWebMvcTest {
         verify(userService, never()).updateFavoritePlatform(anyLong(), any());
     }
 
+    @Test
+    void updatePlatformMails_StoresTheFlagAndRefusesAMissingOne() throws Exception {
+        SecurityContextHolder.getContext().setAuthentication(new UsernamePasswordAuthenticationToken(
+                new UserPrincipal("1", null), null, Collections.emptyList()));
+
+        mockMvc.perform(put("/api/users/settings/platform-mails")
+                        .contentType(MediaType.APPLICATION_JSON).content("{\"enabled\":false}"))
+                .andExpect(status().isNoContent());
+        mockMvc.perform(put("/api/users/settings/platform-mails")
+                        .contentType(MediaType.APPLICATION_JSON).content("{}"))
+                .andExpect(status().isBadRequest());
+
+        verify(userService).updatePlatformMails(1L, false);
+        verify(userService, never()).updatePlatformMails(anyLong(), eq(true));
+    }
+
     @Test
     void updateProfileTiles_StoresTheTiles() throws Exception {
         SecurityContextHolder.getContext().setAuthentication(new UsernamePasswordAuthenticationToken(

src/test/java/auth_service/modules/user/persistence/UserRepositoryTest.java

@@ -251,6 +251,22 @@ class UserRepositoryTest extends AbstractPersistenceTest {
         assertTrue(findByEmail("public@example.com").orElseThrow().getSettings().isPublicProfile());
     }
 
+    @Test
+    void updatePlatformMails_IsKeptByOtherSettingsWrites() {
+        var userEntity = persisted(UserEntity.builder().email("optout@example.com").password("password")
+                .enabled(true).registrationDate(LocalDateTime.now()).build());
+        assertTrue(findByEmail("optout@example.com").orElseThrow().getSettings().isPlatformMails());
+
+        userRepository.update(userEntity.getId(), user -> user.getSettings().setPlatformMails(false));
+        entityManager.flush();
+        entityManager.clear();
+        userRepository.update(userEntity.getId(), user -> user.getSettings().setFavoritePlatform("PS5"));
+        entityManager.flush();
+        entityManager.clear();
+
+        assertFalse(findByEmail("optout@example.com").orElseThrow().getSettings().isPlatformMails());
+    }
+
     @Test
     void updateSettings_KeepsTheAppearance() {
         var userEntity = persisted(UserEntity.builder()

src/test/java/auth_service/modules/user/service/UserServiceTest.java

@@ -195,6 +195,13 @@ class UserServiceTest extends AbstractUnitTest {
         assertEquals("Switch", appliedChanges(1).getFirst().getSettings().getFavoritePlatform());
     }
 
+    @Test
+    void updatePlatformMails_StoresTheFlag() {
+        userService.updatePlatformMails(1L, false);
+
+        org.junit.jupiter.api.Assertions.assertFalse(appliedChanges(1).getFirst().getSettings().isPlatformMails());
+    }
+
     @Test
     void markWelcomeSeen_StampsNow() {
         userService.markWelcomeSeen(1L);