feat(notifications): grouped email digest, new music as a daily summary (#5346)

Nothing is emailed per event. New music (request_completed) goes out at
most once a day, at the summary hour in each user's own timezone, grouped
by artist. Everything else is batched: one email a window after the first
un-emailed item, holding whatever accumulated.

- Migration 0074: notification_email_settings (summary hour, batch window,
  admin-configurable) and user_notification_email_state (batch start, last
  sent, failures and retry_after per user and group). Existing rows are
  stamped emailed so the upgrade sends no backlog.
- The Notifier stamps emailed_at at write time when the recipient's email
  channel is off, so turning email on later doesn't send old items.
- Read rows are never selected. A row is stamped only after the mailer
  accepts, in one transaction with the state, against the read's clock, so
  a coalesced row updated mid-send stays pending.
- A failed send backs off 5m doubling to 6h; SMTP not configured just waits.
- Links come from the public address; without one the email has none.
- The mailer now RFC 2047-encodes subjects and strips line breaks from them.
- Admin → Integrations gains a Notification emails card.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/db/dbq/models.go

@@ -472,6 +472,13 @@ type NetworkSetting struct {
 	PublicUrl        string
 }
 
+type NotificationEmailSetting struct {
+	ID                 bool
+	SummaryHour        int32
+	BatchWindowMinutes int32
+	UpdatedAt          pgtype.Timestamptz
+}
+
 type PasswordReset struct {
 	Token     string
 	UserID    pgtype.UUID
@@ -842,6 +849,15 @@ type UserNotification struct {
 	EmailedAt   pgtype.Timestamptz
 }
 
+type UserNotificationEmailState struct {
+	UserID        pgtype.UUID
+	EmailGroup    string
+	BatchOpenedAt pgtype.Timestamptz
+	LastSentAt    pgtype.Timestamptz
+	Failures      int32
+	RetryAfter    pgtype.Timestamptz
+}
+
 type UserNotificationPref struct {
 	UserID    pgtype.UUID
 	Kind      string

internal/db/dbq/notifications.sql.go

@@ -23,24 +23,50 @@ func (q *Queries) CountUnreadNotifications(ctx context.Context, userID pgtype.UU
 	return count, err
 }
 
+const getNotificationEmailSettings = `-- name: GetNotificationEmailSettings :one
+SELECT id, summary_hour, batch_window_minutes, updated_at FROM notification_email_settings WHERE id = true
+`
+
+func (q *Queries) GetNotificationEmailSettings(ctx context.Context) (NotificationEmailSetting, error) {
+	row := q.db.QueryRow(ctx, getNotificationEmailSettings)
+	var i NotificationEmailSetting
+	err := row.Scan(
+		&i.ID,
+		&i.SummaryHour,
+		&i.BatchWindowMinutes,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
 const insertNotification = `-- name: InsertNotification :one
 
-INSERT INTO user_notifications (user_id, kind, payload)
-VALUES ($1, $2, $3)
+INSERT INTO user_notifications (user_id, kind, payload, emailed_at)
+VALUES ($1, $2, $3,
+        CASE WHEN $4::boolean THEN NULL ELSE now() END)
 RETURNING id
 `
 
 type InsertNotificationParams struct {
-	UserID  pgtype.UUID
-	Kind    string
-	Payload []byte
+	UserID      pgtype.UUID
+	Kind        string
+	Payload     []byte
+	EmailWanted bool
 }
 
 // M489 notifications inbox (#726). Rows are written only through
 // internal/notifications.Notifier, which decides recipients and honours each
 // recipient's inbox preference before it reaches these.
+// email_wanted is the recipient's email channel for this kind as of now. A
+// row nobody wants emailed is stamped at once, so the digest never has to
+// judge it and a later "email on" doesn't send an old backlog.
 func (q *Queries) InsertNotification(ctx context.Context, arg InsertNotificationParams) (pgtype.UUID, error) {
-	row := q.db.QueryRow(ctx, insertNotification, arg.UserID, arg.Kind, arg.Payload)
+	row := q.db.QueryRow(ctx, insertNotification,
+		arg.UserID,
+		arg.Kind,
+		arg.Payload,
+		arg.EmailWanted,
+	)
 	var id pgtype.UUID
 	err := row.Scan(&id)
 	return id, err
@@ -70,6 +96,100 @@ func (q *Queries) ListAdminUserIDs(ctx context.Context) ([]pgtype.UUID, error) {
 	return items, nil
 }
 
+const listEmailPendingNotifications = `-- name: ListEmailPendingNotifications :many
+
+SELECT n.id, n.user_id, n.kind, n.payload, n.created_at,
+       u.email::text AS email, u.username, u.display_name, u.timezone,
+       now()::timestamptz AS read_as_of
+  FROM user_notifications n
+  JOIN users u ON u.id = n.user_id
+ WHERE n.read_at IS NULL
+   AND n.emailed_at IS NULL
+   AND u.email IS NOT NULL AND u.email <> ''
+ ORDER BY n.user_id, n.created_at, n.id
+`
+
+type ListEmailPendingNotificationsRow struct {
+	ID          pgtype.UUID
+	UserID      pgtype.UUID
+	Kind        string
+	Payload     []byte
+	CreatedAt   pgtype.Timestamptz
+	Email       string
+	Username    string
+	DisplayName *string
+	Timezone    string
+	ReadAsOf    pgtype.Timestamptz
+}
+
+// Email digest (#5346) ------------------------------------------------------
+// Every unread, un-emailed row of a user who has an address, oldest first.
+// Read rows are never selected: the user has seen them, so they are not news.
+// read_as_of is the database's clock at the read, handed back to
+// MarkNotificationsEmailed.
+func (q *Queries) ListEmailPendingNotifications(ctx context.Context) ([]ListEmailPendingNotificationsRow, error) {
+	rows, err := q.db.Query(ctx, listEmailPendingNotifications)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []ListEmailPendingNotificationsRow
+	for rows.Next() {
+		var i ListEmailPendingNotificationsRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.UserID,
+			&i.Kind,
+			&i.Payload,
+			&i.CreatedAt,
+			&i.Email,
+			&i.Username,
+			&i.DisplayName,
+			&i.Timezone,
+			&i.ReadAsOf,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const listNotificationEmailState = `-- name: ListNotificationEmailState :many
+SELECT user_id, email_group, batch_opened_at, last_sent_at, failures, retry_after
+  FROM user_notification_email_state
+`
+
+func (q *Queries) ListNotificationEmailState(ctx context.Context) ([]UserNotificationEmailState, error) {
+	rows, err := q.db.Query(ctx, listNotificationEmailState)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []UserNotificationEmailState
+	for rows.Next() {
+		var i UserNotificationEmailState
+		if err := rows.Scan(
+			&i.UserID,
+			&i.EmailGroup,
+			&i.BatchOpenedAt,
+			&i.LastSentAt,
+			&i.Failures,
+			&i.RetryAfter,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const listNotificationPrefsForKind = `-- name: ListNotificationPrefsForKind :many
 SELECT user_id, inbox, phone, email
   FROM user_notification_prefs
@@ -258,6 +378,31 @@ func (q *Queries) MarkNotificationRead(ctx context.Context, arg MarkNotification
 	return result.RowsAffected(), nil
 }
 
+const markNotificationsEmailed = `-- name: MarkNotificationsEmailed :execrows
+UPDATE user_notifications
+   SET emailed_at = now()
+ WHERE id = ANY($1::uuid[])
+   AND created_at <= $2::timestamptz
+   AND emailed_at IS NULL
+`
+
+type MarkNotificationsEmailedParams struct {
+	Ids      []pgtype.UUID
+	ReadAsOf pgtype.Timestamptz
+}
+
+// Stamps the rows an email carried, or that were judged not to need one.
+// read_as_of is from ListEmailPendingNotifications: a coalesced row updated
+// since that read carries a later created_at, holds newer news, and stays
+// pending for the next email.
+func (q *Queries) MarkNotificationsEmailed(ctx context.Context, arg MarkNotificationsEmailedParams) (int64, error) {
+	result, err := q.db.Exec(ctx, markNotificationsEmailed, arg.Ids, arg.ReadAsOf)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected(), nil
+}
+
 const trimNotifications = `-- name: TrimNotifications :execrows
 DELETE FROM user_notifications
  WHERE (read_at IS NOT NULL AND read_at < $1)
@@ -279,20 +424,48 @@ func (q *Queries) TrimNotifications(ctx context.Context, arg TrimNotificationsPa
 	return result.RowsAffected(), nil
 }
 
+const updateNotificationEmailSettings = `-- name: UpdateNotificationEmailSettings :one
+UPDATE notification_email_settings
+   SET summary_hour         = $1,
+       batch_window_minutes = $2,
+       updated_at           = now()
+ WHERE id = true
+RETURNING id, summary_hour, batch_window_minutes, updated_at
+`
+
+type UpdateNotificationEmailSettingsParams struct {
+	SummaryHour        int32
+	BatchWindowMinutes int32
+}
+
+// Migration 0074's CHECKs are the backstop behind the service's validation.
+func (q *Queries) UpdateNotificationEmailSettings(ctx context.Context, arg UpdateNotificationEmailSettingsParams) (NotificationEmailSetting, error) {
+	row := q.db.QueryRow(ctx, updateNotificationEmailSettings, arg.SummaryHour, arg.BatchWindowMinutes)
+	var i NotificationEmailSetting
+	err := row.Scan(
+		&i.ID,
+		&i.SummaryHour,
+		&i.BatchWindowMinutes,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
 const upsertCoalescedNotification = `-- name: UpsertCoalescedNotification :one
-INSERT INTO user_notifications (user_id, kind, payload, coalesce_key)
-VALUES ($1, $2, $3, $4)
+INSERT INTO user_notifications (user_id, kind, payload, coalesce_key, emailed_at)
+VALUES ($1, $2, $3, $4,
+        CASE WHEN $5::boolean THEN NULL ELSE now() END)
 ON CONFLICT (user_id, coalesce_key) WHERE read_at IS NULL AND coalesce_key IS NOT NULL
 DO UPDATE SET
     payload = CASE
-        WHEN $5::boolean THEN
+        WHEN $6::boolean THEN
             EXCLUDED.payload || jsonb_build_object('count',
                 COALESCE((user_notifications.payload->>'count')::bigint, 0)
               + COALESCE((EXCLUDED.payload->>'count')::bigint, 0))
         ELSE EXCLUDED.payload
     END,
     created_at = now(),
-    emailed_at = NULL
+    emailed_at = EXCLUDED.emailed_at
 RETURNING id
 `
 
@@ -301,6 +474,7 @@ type UpsertCoalescedNotificationParams struct {
 	Kind        string
 	Payload     []byte
 	CoalesceKey *string
+	EmailWanted bool
 	SumCount    bool
 }
 
@@ -314,12 +488,14 @@ type UpsertCoalescedNotificationParams struct {
 //
 // emailed_at is cleared so the newer state goes out in the next batch; an
 // emailed-but-unread row that keeps growing is news the user hasn't seen.
+// Unless email_wanted is false, as in InsertNotification.
 func (q *Queries) UpsertCoalescedNotification(ctx context.Context, arg UpsertCoalescedNotificationParams) (pgtype.UUID, error) {
 	row := q.db.QueryRow(ctx, upsertCoalescedNotification,
 		arg.UserID,
 		arg.Kind,
 		arg.Payload,
 		arg.CoalesceKey,
+		arg.EmailWanted,
 		arg.SumCount,
 	)
 	var id pgtype.UUID
@@ -327,6 +503,39 @@ func (q *Queries) UpsertCoalescedNotification(ctx context.Context, arg UpsertCoa
 	return id, err
 }
 
+const upsertNotificationEmailState = `-- name: UpsertNotificationEmailState :exec
+INSERT INTO user_notification_email_state
+       (user_id, email_group, batch_opened_at, last_sent_at, failures, retry_after)
+VALUES ($1, $2, $3,
+        $4, $5, $6)
+ON CONFLICT (user_id, email_group) DO UPDATE SET
+    batch_opened_at = EXCLUDED.batch_opened_at,
+    last_sent_at    = EXCLUDED.last_sent_at,
+    failures        = EXCLUDED.failures,
+    retry_after     = EXCLUDED.retry_after
+`
+
+type UpsertNotificationEmailStateParams struct {
+	UserID        pgtype.UUID
+	EmailGroup    string
+	BatchOpenedAt pgtype.Timestamptz
+	LastSentAt    pgtype.Timestamptz
+	Failures      int32
+	RetryAfter    pgtype.Timestamptz
+}
+
+func (q *Queries) UpsertNotificationEmailState(ctx context.Context, arg UpsertNotificationEmailStateParams) error {
+	_, err := q.db.Exec(ctx, upsertNotificationEmailState,
+		arg.UserID,
+		arg.EmailGroup,
+		arg.BatchOpenedAt,
+		arg.LastSentAt,
+		arg.Failures,
+		arg.RetryAfter,
+	)
+	return err
+}
+
 const upsertNotificationPref = `-- name: UpsertNotificationPref :exec
 INSERT INTO user_notification_prefs (user_id, kind, inbox, phone, email)
 VALUES ($1, $2, $3, $4, $5)