feat(notifications): Discord bot DM channel with account linking

Adds Discord direct messages as a notification channel. Users link their
Discord account via OAuth2 (identify scope only, one-time server-side
state rows); a bot delivers their inbox notifications as DMs.

- Extract the email channel's watermark sweep into a generic
  account-channel engine; email and Discord are now thin adapters, so
  the SKIP LOCKED claim / watermark-after-send durability logic exists
  once.
- New internal/discord REST client (token exchange, identity, open DM,
  send message) — no Gateway connection, no new dependencies.
- Opt-in master switch (notifications.discord_enabled, default off)
  gates delivery, linking, capability, and the admin settings reveal.
- Admin UI: credentials (secret + bot token encrypted at rest), dev
  portal setup checklist, bot invite link buttons, and a test button
  that bypasses the settings read cache and is disabled while
  credential edits are unsaved.
- DM failures from missing shared guild (Discord 50007) surface as link
  health in user settings and self-heal via capped backoff.
- New combined mode (per_episode_and_digest) for email and Discord:
  instant sends all day plus a daily digest recapping the whole window
  since the previous digest.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Quick
2026-06-11 19:51:03 -04:00
co-authored by Claude Fable 5
parent 27b9006ef5
commit 5f05374a1d
25 changed files with 2496 additions and 406 deletions
+3
View File
@@ -59,6 +59,9 @@ POSTGRES_SHM_SIZE=8gb
# docs/architecture/secret-encryption.md.
# SECRET_KEY=replace-with-output-of-openssl-rand-base64-48
# Optional public URL for Silo. If not set, the server will use the IP address of the container.
# SILO_PUBLIC_URL=https://silo.example.com
# Run from source / advanced overrides
# Only DATABASE_URL is required when running Silo outside the default docker compose stack.
# DATABASE_URL=postgres://silo:password@localhost:5432/silo
+31 -11
View File
@@ -277,15 +277,18 @@ func (h *NotificationsHandler) HandleUpdatePreferences(w http.ResponseWriter, r
}
type capabilityResponse struct {
InApp capabilityInApp `json:"in_app"`
ApplePush capabilityPush `json:"apple_push"`
AndroidPush capabilityPush `json:"android_push"`
WebPush capabilityWebPush `json:"web_push"`
Webhooks capabilityWebhooks `json:"webhooks"`
Email capabilityEmail `json:"email"`
InApp capabilityInApp `json:"in_app"`
ApplePush capabilityPush `json:"apple_push"`
AndroidPush capabilityPush `json:"android_push"`
WebPush capabilityWebPush `json:"web_push"`
Webhooks capabilityWebhooks `json:"webhooks"`
Email capabilityAccountChannel `json:"email"`
Discord capabilityAccountChannel `json:"discord"`
}
type capabilityEmail struct {
// capabilityAccountChannel describes an account-level digest channel (email,
// Discord DMs).
type capabilityAccountChannel struct {
Available bool `json:"available"`
// Modes lists the cadences users may pick (per-episode is an admin
// allowance); DigestHour tells the UI when daily digests go out.
@@ -333,18 +336,34 @@ func (h *NotificationsHandler) HandleCapability(w http.ResponseWriter, r *http.R
webPush = capabilityWebPush{Available: true, PublicKey: publicKey}
}
}
email := capabilityEmail{Modes: []string{}}
email := capabilityAccountChannel{Modes: []string{}}
if h.system.EmailAvailable(r.Context()) {
modes := []string{notifications.EmailModeDailyDigest}
modes := []string{notifications.ChannelModeDailyDigest}
if h.system.Settings.EmailAllowPerEpisode(r.Context()) {
modes = append(modes, notifications.EmailModePerEpisode)
modes = append(modes,
notifications.ChannelModePerEpisode,
notifications.ChannelModePerEpisodeAndDigest)
}
email = capabilityEmail{
email = capabilityAccountChannel{
Available: true,
Modes: modes,
DigestHour: h.system.Settings.EmailDigestHour(r.Context()),
}
}
discordCap := capabilityAccountChannel{Modes: []string{}}
if h.system.DiscordAvailable(r.Context()) {
modes := []string{notifications.ChannelModeDailyDigest}
if h.system.Settings.DiscordAllowPerEpisode(r.Context()) {
modes = append(modes,
notifications.ChannelModePerEpisode,
notifications.ChannelModePerEpisodeAndDigest)
}
discordCap = capabilityAccountChannel{
Available: true,
Modes: modes,
DigestHour: h.system.Settings.DiscordDigestHour(r.Context()),
}
}
writeJSON(w, http.StatusOK, capabilityResponse{
InApp: capabilityInApp{Enabled: h.system.Settings.UIEnabled(r.Context())},
ApplePush: capabilityPush{Available: false, Provider: "off", SupportedModes: []string{"in_app_only"}},
@@ -352,6 +371,7 @@ func (h *NotificationsHandler) HandleCapability(w http.ResponseWriter, r *http.R
WebPush: webPush,
Webhooks: webhooks,
Email: email,
Discord: discordCap,
})
}
@@ -0,0 +1,230 @@
package handlers
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"net/http"
"net/url"
"strings"
"time"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/discord"
"github.com/Silo-Server/silo-server/internal/notifications"
)
// discordSettingsPath is the SPA page the OAuth callback redirects back to,
// with ?discord_linked=1 or ?discord_error=<reason> for the page to toast.
const discordSettingsPath = "/settings/notifications"
// DiscordNotificationsHandler serves the account-level Discord DM channel:
// preferences, the OAuth account-link flow, and the admin bot test. The
// preferences and link-init endpoints require auth; the link callback is
// public because Discord redirects the browser there without credentials —
// the one-time server-side state row authenticates it instead.
type DiscordNotificationsHandler struct {
system *notifications.System
publicURL string
}
// NewDiscordNotificationsHandler creates a DiscordNotificationsHandler.
// publicURL may be empty, in which case linking is reported unavailable
// (Discord needs a stable redirect_uri origin).
func NewDiscordNotificationsHandler(system *notifications.System, publicURL string) *DiscordNotificationsHandler {
return &DiscordNotificationsHandler{system: system, publicURL: strings.TrimRight(publicURL, "/")}
}
func (h *DiscordNotificationsHandler) redirectURI() string {
return h.publicURL + "/api/v1/notifications/discord/link/callback"
}
// discordPreferencesResponse is the account-level Discord DM setting plus
// link state and health.
type discordPreferencesResponse struct {
Linked bool `json:"linked"`
DiscordUsername string `json:"discord_username,omitempty"`
Mode string `json:"mode"`
LinkFailure string `json:"link_failure,omitempty"`
}
type updateDiscordPreferencesRequest struct {
Mode string `json:"mode"`
}
func discordPrefsResponse(prefs notifications.DiscordPrefs) discordPreferencesResponse {
return discordPreferencesResponse{
Linked: prefs.Linked(),
DiscordUsername: prefs.DiscordUsername,
Mode: prefs.Mode,
LinkFailure: prefs.LinkFailure,
}
}
// HandleGetPreferences handles GET /notifications/discord-preferences.
func (h *DiscordNotificationsHandler) HandleGetPreferences(w http.ResponseWriter, r *http.Request) {
userID := apimw.GetUserID(r.Context())
prefs, err := h.system.DiscordPrefsFor(r.Context(), userID)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load Discord preferences")
return
}
writeJSON(w, http.StatusOK, discordPrefsResponse(prefs))
}
// HandleUpdatePreferences handles PUT /notifications/discord-preferences.
func (h *DiscordNotificationsHandler) HandleUpdatePreferences(w http.ResponseWriter, r *http.Request) {
userID := apimw.GetUserID(r.Context())
var req updateDiscordPreferencesRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
err := h.system.SetDiscordMode(r.Context(), userID, req.Mode)
switch {
case err == nil:
case errors.Is(err, notifications.ErrDiscordModeInvalid):
writeError(w, http.StatusBadRequest, "bad_request", "Unknown Discord notification mode")
return
case errors.Is(err, notifications.ErrDiscordModeNotAllowed):
writeError(w, http.StatusBadRequest, "not_allowed", "Per-episode Discord DMs are disabled by the administrator")
return
case errors.Is(err, notifications.ErrDiscordNotLinked):
writeError(w, http.StatusBadRequest, "not_linked", "Link a Discord account first")
return
default:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to save Discord preferences")
return
}
prefs, err := h.system.DiscordPrefsFor(r.Context(), userID)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load Discord preferences")
return
}
writeJSON(w, http.StatusOK, discordPrefsResponse(prefs))
}
// HandleUnlink handles DELETE /notifications/discord-link: removes the
// linked identity and switches the channel off.
func (h *DiscordNotificationsHandler) HandleUnlink(w http.ResponseWriter, r *http.Request) {
userID := apimw.GetUserID(r.Context())
if err := h.system.UnlinkDiscord(r.Context(), userID); err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to unlink Discord account")
return
}
writeJSON(w, http.StatusOK, discordPreferencesResponse{Mode: notifications.ChannelModeOff})
}
type discordLinkInitResponse struct {
URL string `json:"url"`
}
// HandleLinkInit handles POST /notifications/discord/link/init: records a
// one-time state for the signed-in account and returns the Discord consent
// URL for the SPA to navigate to.
func (h *DiscordNotificationsHandler) HandleLinkInit(w http.ResponseWriter, r *http.Request) {
userID := apimw.GetUserID(r.Context())
if !h.system.DiscordAvailable(r.Context()) {
writeError(w, http.StatusConflict, "not_configured", "Discord integration is not enabled by the administrator")
return
}
if h.publicURL == "" {
writeError(w, http.StatusConflict, "no_public_url", "Linking requires SILO_PUBLIC_URL to be configured")
return
}
stateBytes := make([]byte, 32)
if _, err := rand.Read(stateBytes); err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to start Discord link")
return
}
state := hex.EncodeToString(stateBytes)
if err := h.system.BeginDiscordLink(r.Context(), state, userID); err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to start Discord link")
return
}
query := url.Values{
"client_id": {h.system.Settings.DiscordClientID(r.Context())},
"response_type": {"code"},
"scope": {"identify"},
"redirect_uri": {h.redirectURI()},
"state": {state},
}
writeJSON(w, http.StatusOK, discordLinkInitResponse{
URL: discord.AuthorizeURL + "?" + query.Encode(),
})
}
// HandleLinkCallback handles GET /notifications/discord/link/callback.
// Public route: Discord redirects the user's browser here. The one-time
// state row authenticates the request and recovers which account started
// the flow; the browser is then sent back to the settings page.
func (h *DiscordNotificationsHandler) HandleLinkCallback(w http.ResponseWriter, r *http.Request) {
redirectBack := func(params url.Values) {
http.Redirect(w, r, discordSettingsPath+"?"+params.Encode(), http.StatusFound)
}
// The admin may have switched the integration off between init and
// callback; don't complete a link for a disabled channel.
if !h.system.DiscordAvailable(r.Context()) {
redirectBack(url.Values{"discord_error": {"disabled"}})
return
}
query := r.URL.Query()
if query.Get("error") != "" {
// The user declined on Discord's consent screen.
redirectBack(url.Values{"discord_error": {"denied"}})
return
}
state := query.Get("state")
code := query.Get("code")
if state == "" || code == "" {
redirectBack(url.Values{"discord_error": {"invalid_callback"}})
return
}
userID, ok, err := h.system.ConsumeDiscordLinkState(r.Context(), state)
if err != nil || !ok {
redirectBack(url.Values{"discord_error": {"state_invalid"}})
return
}
if _, err := h.system.CompleteDiscordLink(r.Context(), userID, code, h.redirectURI()); err != nil {
redirectBack(url.Values{"discord_error": {"exchange_failed"}})
return
}
redirectBack(url.Values{"discord_linked": {"1"}})
}
type discordTestResponse struct {
OK bool `json:"ok"`
DurationMS int64 `json:"duration_ms"`
Message string `json:"message"`
}
// HandleAdminTest handles POST /admin/notifications/discord/test: verifies
// the configured bot token by fetching the bot's own identity.
func (h *DiscordNotificationsHandler) HandleAdminTest(w http.ResponseWriter, r *http.Request) {
start := time.Now()
botUser, err := h.system.TestDiscordBot(r.Context())
duration := time.Since(start).Milliseconds()
if errors.Is(err, notifications.ErrDiscordNotConfigured) {
writeJSON(w, http.StatusOK, discordTestResponse{
OK: false, DurationMS: duration, Message: "Bot token is not configured",
})
return
}
if err != nil {
writeJSON(w, http.StatusOK, discordTestResponse{
OK: false, DurationMS: duration, Message: err.Error(),
})
return
}
writeJSON(w, http.StatusOK, discordTestResponse{
OK: true, DurationMS: duration, Message: "Connected as " + botUser.Username,
})
}
+20
View File
@@ -1479,6 +1479,17 @@ func NewRouter(deps Dependencies) chi.Router {
})
}
// Discord account-link OAuth callback: public — Discord redirects the
// browser here without credentials; the one-time link-state row
// authenticates the request and maps it back to the initiating
// account. The static path coexists with the authenticated
// /notifications subrouter below (static routes win in chi).
var discordNotificationsHandler *handlers.DiscordNotificationsHandler
if deps.Notifications != nil {
discordNotificationsHandler = handlers.NewDiscordNotificationsHandler(deps.Notifications, deps.PublicURL)
r.Get("/notifications/discord/link/callback", discordNotificationsHandler.HandleLinkCallback)
}
// API key management routes (auth only, no viewer access needed).
if apiKeyRepo != nil && authMiddleware != nil {
r.Group(func(r chi.Router) {
@@ -1546,6 +1557,12 @@ func NewRouter(deps Dependencies) chi.Router {
r.Put("/preferences", notificationsHandler.HandleUpdatePreferences)
r.Get("/email-preferences", notificationsHandler.HandleGetEmailPreferences)
r.Put("/email-preferences", notificationsHandler.HandleUpdateEmailPreferences)
if discordNotificationsHandler != nil {
r.Get("/discord-preferences", discordNotificationsHandler.HandleGetPreferences)
r.Put("/discord-preferences", discordNotificationsHandler.HandleUpdatePreferences)
r.Delete("/discord-link", discordNotificationsHandler.HandleUnlink)
r.Post("/discord/link/init", discordNotificationsHandler.HandleLinkInit)
}
r.Post("/read-all", notificationsHandler.HandleReadAll)
r.Route("/webhooks", func(r chi.Router) {
r.Get("/", notificationsHandler.HandleListWebhooks)
@@ -2163,6 +2180,9 @@ func NewRouter(deps Dependencies) chi.Router {
emailHandler := handlers.NewEmailHandler(mail.NewSMTPSender(settingsRepo))
r.Post("/email/test", emailHandler.HandleTest)
}
if discordNotificationsHandler != nil {
r.Post("/notifications/discord/test", discordNotificationsHandler.HandleAdminTest)
}
if adminIntroHandler != nil {
r.Post("/items/{id}/refresh-markers", adminIntroHandler.HandleRefreshEpisodeMarkers)
r.Post("/items/{id}/redetect-intro", adminIntroHandler.HandleRedetectEpisodeIntro)
@@ -95,6 +95,11 @@ var SensitiveSettingKeys = map[string]bool{
// Shared outbound email (internal/mail) SMTP credential.
"email.smtp_password": true,
// Discord notification integration. The client_id is public in Discord's
// own UI, so only the secret and bot token are encrypted.
"discord.client_secret": true,
"discord.bot_token": true,
// Web Push VAPID keypair JSON (generated + persisted atomically as one
// value by the notifications system; clients receive the public half via
// the capability endpoint, never from the settings store).
+226
View File
@@ -0,0 +1,226 @@
// Package discord is a minimal Discord REST client covering exactly what the
// notification channel needs: the OAuth2 code exchange used for account
// linking, identity lookups, and bot DM delivery. No Gateway connection is
// held; everything is short-lived REST against a fixed trusted host.
package discord
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
"golang.org/x/time/rate"
)
const (
defaultAPIBase = "https://discord.com/api/v10"
// AuthorizeURL is the user-facing OAuth2 consent page.
AuthorizeURL = "https://discord.com/oauth2/authorize"
requestTimeout = 10 * time.Second
// Discord asks bot user agents to identify themselves in this format.
userAgent = "DiscordBot (https://github.com/Silo-Server/silo-server, 1.0)"
// errorBodyLimit bounds how much of an error response is read for
// diagnostics.
errorBodyLimit = 4 << 10
)
// Sentinel errors for the failure modes callers branch on.
var (
// ErrDMBlocked is Discord error 50007: the bot cannot DM this user. The
// user does not share a guild with the bot or has server DMs disabled.
ErrDMBlocked = errors.New("discord: cannot send messages to this user")
// ErrUnauthorized means the bot token or OAuth credentials were rejected.
ErrUnauthorized = errors.New("discord: unauthorized")
// ErrRateLimited means Discord returned 429; retry later.
ErrRateLimited = errors.New("discord: rate limited")
)
const dmBlockedCode = 50007
// User is the subset of a Discord user object the integration stores.
type User struct {
ID string `json:"id"`
Username string `json:"username"`
}
// Client makes Discord REST calls. Tokens are passed per call so callers can
// read live settings; the client itself holds no credentials.
type Client struct {
httpClient *http.Client
apiBase string
// limiter paces outbound calls well under Discord's global rate limits
// (~50 req/s global, ~5 DMs/s); notification volume is far below this,
// but digest-hour bursts across many accounts need smoothing.
limiter *rate.Limiter
}
// NewClient creates a Client.
func NewClient() *Client {
return &Client{
httpClient: &http.Client{Timeout: requestTimeout},
apiBase: defaultAPIBase,
limiter: rate.NewLimiter(rate.Every(250*time.Millisecond), 4),
}
}
// ExchangeCode performs the OAuth2 authorization-code exchange and returns
// the user access token. Only the identify scope is requested at authorize
// time, so the token can do nothing beyond reading the user's own identity.
func (c *Client) ExchangeCode(ctx context.Context, clientID, clientSecret, code, redirectURI string) (string, error) {
form := url.Values{
"grant_type": {"authorization_code"},
"code": {code},
"redirect_uri": {redirectURI},
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
c.apiBase+"/oauth2/token", strings.NewReader(form.Encode()))
if err != nil {
return "", err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.SetBasicAuth(clientID, clientSecret)
var token struct {
AccessToken string `json:"access_token"`
}
if err := c.do(req, &token); err != nil {
return "", fmt.Errorf("exchange oauth code: %w", err)
}
if token.AccessToken == "" {
return "", errors.New("discord: token response missing access_token")
}
return token.AccessToken, nil
}
// GetUser returns the identity behind a user access token (GET /users/@me).
func (c *Client) GetUser(ctx context.Context, accessToken string) (User, error) {
return c.getMe(ctx, "Bearer "+accessToken)
}
// GetBotUser returns the bot's own identity, verifying the bot token. Used by
// the admin "test" endpoint.
func (c *Client) GetBotUser(ctx context.Context, botToken string) (User, error) {
return c.getMe(ctx, "Bot "+botToken)
}
func (c *Client) getMe(ctx context.Context, authorization string) (User, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.apiBase+"/users/@me", nil)
if err != nil {
return User{}, err
}
req.Header.Set("Authorization", authorization)
var user User
if err := c.do(req, &user); err != nil {
return User{}, fmt.Errorf("get current user: %w", err)
}
if user.ID == "" {
return User{}, errors.New("discord: user response missing id")
}
return user, nil
}
// OpenDMChannel opens (or returns the existing) DM channel with the user.
// Idempotent: Discord returns the same channel for repeated calls.
func (c *Client) OpenDMChannel(ctx context.Context, botToken, recipientDiscordUserID string) (string, error) {
body, err := json.Marshal(map[string]string{"recipient_id": recipientDiscordUserID})
if err != nil {
return "", err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
c.apiBase+"/users/@me/channels", bytes.NewReader(body))
if err != nil {
return "", err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bot "+botToken)
var channel struct {
ID string `json:"id"`
}
if err := c.do(req, &channel); err != nil {
return "", fmt.Errorf("open dm channel: %w", err)
}
if channel.ID == "" {
return "", errors.New("discord: channel response missing id")
}
return channel.ID, nil
}
// SendDM posts a message payload (Discord message JSON, e.g. an embeds body)
// to a DM channel.
func (c *Client) SendDM(ctx context.Context, botToken, channelID string, payload []byte) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
c.apiBase+"/channels/"+url.PathEscape(channelID)+"/messages", bytes.NewReader(payload))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bot "+botToken)
if err := c.do(req, nil); err != nil {
return fmt.Errorf("send dm: %w", err)
}
return nil
}
// do executes the request under the rate limiter and decodes a 2xx JSON
// response into out (when non-nil). Non-2xx responses map to sentinel errors
// with the Discord error message attached.
func (c *Client) do(req *http.Request, out any) error {
if err := c.limiter.Wait(req.Context()); err != nil {
return err
}
req.Header.Set("User-Agent", userAgent)
resp, err := c.httpClient.Do(req)
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return apiError(resp)
}
if out == nil {
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, errorBodyLimit))
return nil
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(out); err != nil {
return fmt.Errorf("decode discord response: %w", err)
}
return nil
}
// apiError maps a non-2xx response to a sentinel error, preserving Discord's
// message for logs and UI surfacing.
func apiError(resp *http.Response) error {
body, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit))
var payload struct {
Code int `json:"code"`
Message string `json:"message"`
}
_ = json.Unmarshal(body, &payload)
switch {
case payload.Code == dmBlockedCode:
return ErrDMBlocked
case resp.StatusCode == http.StatusUnauthorized:
return fmt.Errorf("%w: %s", ErrUnauthorized, payload.Message)
case resp.StatusCode == http.StatusTooManyRequests:
return ErrRateLimited
}
message := payload.Message
if message == "" {
message = strings.TrimSpace(string(body))
}
return fmt.Errorf("discord: HTTP %d: %s", resp.StatusCode, message)
}
+136
View File
@@ -0,0 +1,136 @@
package discord
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"testing"
"time"
"golang.org/x/time/rate"
)
func testClient(t *testing.T, handler http.Handler) *Client {
t.Helper()
server := httptest.NewServer(handler)
t.Cleanup(server.Close)
return &Client{
httpClient: server.Client(),
apiBase: server.URL,
limiter: rate.NewLimiter(rate.Inf, 1),
}
}
func TestExchangeCodeAndGetUser(t *testing.T) {
client := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/oauth2/token":
user, pass, ok := r.BasicAuth()
if !ok || user != "client-id" || pass != "client-secret" {
t.Errorf("missing or wrong basic auth: %s/%s", user, pass)
}
if err := r.ParseForm(); err != nil {
t.Fatal(err)
}
if r.PostForm.Get("grant_type") != "authorization_code" || r.PostForm.Get("code") != "the-code" {
t.Errorf("unexpected form: %v", r.PostForm)
}
_ = json.NewEncoder(w).Encode(map[string]string{"access_token": "user-token"})
case "/users/@me":
if got := r.Header.Get("Authorization"); got != "Bearer user-token" {
t.Errorf("unexpected authorization %q", got)
}
_ = json.NewEncoder(w).Encode(User{ID: "42", Username: "quick"})
default:
t.Errorf("unexpected path %s", r.URL.Path)
w.WriteHeader(http.StatusNotFound)
}
}))
token, err := client.ExchangeCode(context.Background(), "client-id", "client-secret", "the-code", "https://silo.example/cb")
if err != nil {
t.Fatalf("exchange: %v", err)
}
user, err := client.GetUser(context.Background(), token)
if err != nil {
t.Fatalf("get user: %v", err)
}
if user.ID != "42" || user.Username != "quick" {
t.Fatalf("unexpected user %+v", user)
}
}
func TestOpenDMChannelAndSendDM(t *testing.T) {
client := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/users/@me/channels":
if got := r.Header.Get("Authorization"); got != "Bot bot-token" {
t.Errorf("unexpected authorization %q", got)
}
var body map[string]string
_ = json.NewDecoder(r.Body).Decode(&body)
if body["recipient_id"] != "42" {
t.Errorf("unexpected recipient %q", body["recipient_id"])
}
_ = json.NewEncoder(w).Encode(map[string]string{"id": "dm-123"})
case "/channels/dm-123/messages":
w.WriteHeader(http.StatusOK)
default:
t.Errorf("unexpected path %s", r.URL.Path)
w.WriteHeader(http.StatusNotFound)
}
}))
channelID, err := client.OpenDMChannel(context.Background(), "bot-token", "42")
if err != nil {
t.Fatalf("open dm: %v", err)
}
if channelID != "dm-123" {
t.Fatalf("unexpected channel id %q", channelID)
}
if err := client.SendDM(context.Background(), "bot-token", channelID, []byte(`{"content":"hi"}`)); err != nil {
t.Fatalf("send dm: %v", err)
}
}
func TestErrorMapping(t *testing.T) {
cases := []struct {
name string
status int
body string
want error
}{
{"dm blocked", http.StatusForbidden, `{"code":50007,"message":"Cannot send messages to this user"}`, ErrDMBlocked},
{"bad token", http.StatusUnauthorized, `{"message":"401: Unauthorized"}`, ErrUnauthorized},
{"rate limited", http.StatusTooManyRequests, `{"message":"You are being rate limited."}`, ErrRateLimited},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
client := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(tc.status)
_, _ = w.Write([]byte(tc.body))
}))
err := client.SendDM(context.Background(), "bot-token", "dm-123", []byte(`{}`))
if !errors.Is(err, tc.want) {
t.Fatalf("got %v, want %v", err, tc.want)
}
})
}
}
func TestLimiterHonorsContextCancel(t *testing.T) {
client := testClient(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
}))
// Exhausted limiter with a long refill forces Wait to block on the context.
client.limiter = rate.NewLimiter(rate.Every(time.Hour), 1)
client.limiter.Allow()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
defer cancel()
if err := client.SendDM(ctx, "bot-token", "dm-123", []byte(`{}`)); err == nil {
t.Fatal("expected context error from limiter wait")
}
}
@@ -0,0 +1,418 @@
package notifications
import (
"context"
"errors"
"fmt"
"log/slog"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// Account-channel modes shared by every account-level notification channel
// (email, Discord). These channels are account-scoped: one setting covers
// every profile on the login account, and the sweep collapses cross-profile
// duplicates into a single send.
const (
ChannelModeOff = "off"
ChannelModePerEpisode = "per_episode"
ChannelModeDailyDigest = "daily_digest"
// ChannelModePerEpisodeAndDigest sends per-episode all day and, at the
// digest hour, a digest recapping everything since the previous digest —
// including items already sent individually.
ChannelModePerEpisodeAndDigest = "per_episode_and_digest"
)
// ValidChannelMode reports whether mode is a recognized account-channel mode.
func ValidChannelMode(mode string) bool {
switch mode {
case ChannelModeOff, ChannelModePerEpisode, ChannelModeDailyDigest,
ChannelModePerEpisodeAndDigest:
return true
}
return false
}
// ModeIncludesPerEpisode reports whether the mode performs per-episode sends
// and is therefore subject to the admin per-episode allowance.
func ModeIncludesPerEpisode(mode string) bool {
return mode == ChannelModePerEpisode || mode == ChannelModePerEpisodeAndDigest
}
const (
channelPollInterval = time.Minute
// channelNudgeDelay coalesces the per-row dispatch nudges of one fanout
// batch (all rows commit before the first nudge fires) into one pass.
channelNudgeDelay = 2 * time.Second
// channelFetchLimit bounds one send's worth of watermark progress; the
// next pass drains the remainder.
channelFetchLimit = 200
// channelMaxFailuresPerPass stops a pass early when sends keep failing —
// transport trouble is almost always global, not per-recipient.
channelMaxFailuresPerPass = 3
channelFailureBackoffBase = time.Minute
channelFailureBackoffMax = 6 * time.Hour
)
// errChannelUnavailable aborts a sweep pass entirely: the channel's transport
// is unconfigured or globally down, so nothing else will send either. The
// failing account is not penalized with backoff.
var errChannelUnavailable = errors.New("notification channel unavailable")
// effectiveChannelMode coerces per-episode modes to the daily digest when the
// admin has disallowed per-episode sends, instead of silencing those accounts.
func effectiveChannelMode(mode string, allowPerEpisode bool) string {
if ModeIncludesPerEpisode(mode) && !allowPerEpisode {
return ChannelModeDailyDigest
}
return mode
}
// cursorLess orders cursors the way the delivery queries do: by
// (created_at, id).
func cursorLess(a, b Cursor) bool {
if !a.CreatedAt.Equal(b.CreatedAt) {
return a.CreatedAt.Before(b.CreatedAt)
}
return a.ID < b.ID
}
// channelDigestDue reports whether a daily digest should go out: today's send
// time (digestHour, local) has passed and no digest was stamped since.
func channelDigestDue(now time.Time, digestHour int, lastDigestAt *time.Time) bool {
todaySend := time.Date(now.Year(), now.Month(), now.Day(), digestHour, 0, 0, 0, now.Location())
if now.Before(todaySend) {
return false
}
return lastDigestAt == nil || lastDigestAt.Before(todaySend)
}
// channelRetryEligible applies exponential backoff after failed sends:
// 1m, 2m, 4m, ... capped at channelFailureBackoffMax.
func channelRetryEligible(now time.Time, lastAttemptAt *time.Time, consecutiveFailures int) bool {
if consecutiveFailures <= 0 || lastAttemptAt == nil {
return true
}
backoff := channelFailureBackoffBase << min(consecutiveFailures-1, 30)
if backoff > channelFailureBackoffMax || backoff <= 0 {
backoff = channelFailureBackoffMax
}
return !now.Before(lastAttemptAt.Add(backoff))
}
// accountRecipient is the channel-agnostic sweep state for one account: the
// user-chosen mode plus the dispatch watermark and failure backoff counters.
// Channel-specific contact details (email address, Discord identity) stay
// inside the channel implementation.
type accountRecipient struct {
UserID int
Mode string
WatermarkCreatedAt time.Time
WatermarkID string
LastDigestAt *time.Time
LastAttemptAt *time.Time
ConsecutiveFailures int
}
// accountChannel supplies the channel-specific pieces of the account
// watermark sweep: prefs-table access and the actual send. The engine owns
// the loop, eligibility, claim transaction, and watermark advancement.
type accountChannel interface {
// name labels log lines.
name() string
// enabled gates a whole pass (kill switch + transport configured).
enabled(ctx context.Context) bool
// allowPerEpisode is the admin allowance for per-send mode.
allowPerEpisode(ctx context.Context) bool
// digestHour is the hour of day (0-23, server-local) for daily digests.
digestHour(ctx context.Context) int
// listRecipients returns every account with the channel switched on and a
// usable destination. Disabled or deleted accounts must not appear.
listRecipients(ctx context.Context) ([]accountRecipient, error)
// claim locks the account's prefs row for one dispatch attempt with
// FOR UPDATE SKIP LOCKED; (nil, nil) means another node holds the row.
claim(ctx context.Context, tx pgx.Tx, userID int) (*accountRecipient, error)
// markSent advances the watermark past everything the send covered and
// resets failure backoff. digestAt is non-nil for digest sends.
markSent(ctx context.Context, tx pgx.Tx, userID int, watermark Cursor, digestAt *time.Time) error
// markFailure records a failed send for backoff; the watermark stays put
// so the next eligible pass retries the same items.
markFailure(ctx context.Context, tx pgx.Tx, userID int, sendErr error) error
// send delivers one account's pending rows. It runs inside the claim
// transaction; tx is for channel-state updates only (the engine owns
// commit/rollback). Errors wrapping errChannelUnavailable abort the pass
// without penalizing the account.
send(ctx context.Context, tx pgx.Tx, userID int, mode string, rows []DeliveryRow) error
}
// accountChannelWorker drives one account-level channel. Unlike webhooks and
// web push it keeps no per-target outbox: deliveries already carry user_id,
// so a per-account watermark over notification_deliveries is the durable
// dispatch state. The watermark advances only after a successful send, and
// one send covers everything since the last one — which also collapses the
// duplicate rows an account gets when several of its profiles follow the
// same series.
type accountChannelWorker struct {
pool *pgxpool.Pool
deliveries *DeliveryRepository
channel accountChannel
logger *slog.Logger
nudge chan struct{}
now func() time.Time
}
func newAccountChannelWorker(
pool *pgxpool.Pool,
deliveries *DeliveryRepository,
channel accountChannel,
) *accountChannelWorker {
return &accountChannelWorker{
pool: pool,
deliveries: deliveries,
channel: channel,
logger: slog.Default().With("component", "notifications."+channel.name()),
nudge: make(chan struct{}, 1),
now: time.Now,
}
}
// Nudge schedules a near-term pass so per-episode sends follow fanout within
// seconds instead of waiting for the next poll. Non-blocking.
func (w *accountChannelWorker) Nudge() {
if w == nil {
return
}
select {
case w.nudge <- struct{}{}:
default:
}
}
// Run sweeps eligible accounts until ctx is canceled.
func (w *accountChannelWorker) Run(ctx context.Context) {
ticker := time.NewTicker(channelPollInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
case <-w.nudge:
select {
case <-ctx.Done():
return
case <-time.After(channelNudgeDelay):
}
}
if !w.channel.enabled(ctx) {
continue
}
w.runPass(ctx)
}
}
// runPass attempts one send per eligible account. Failures back off per
// account; the pass aborts entirely on errChannelUnavailable or after a few
// consecutive failures, since both indicate a global transport problem.
func (w *accountChannelWorker) runPass(ctx context.Context) {
recipients, err := w.channel.listRecipients(ctx)
if err != nil {
w.logger.Error("channel pass: list recipients failed", "error", err)
return
}
if len(recipients) == 0 {
return
}
allowPerEpisode := w.channel.allowPerEpisode(ctx)
digestHour := w.channel.digestHour(ctx)
now := w.now()
failures := 0
for _, rec := range recipients {
if ctx.Err() != nil || failures >= channelMaxFailuresPerPass {
return
}
if !channelRetryEligible(now, rec.LastAttemptAt, rec.ConsecutiveFailures) {
continue
}
mode := effectiveChannelMode(rec.Mode, allowPerEpisode)
digestDue := channelDigestDue(now, digestHour, rec.LastDigestAt)
switch mode {
case ChannelModePerEpisode, ChannelModePerEpisodeAndDigest:
if mode == ChannelModePerEpisodeAndDigest && digestDue {
break // the digest leg has work regardless of pending rows
}
// Cheap pre-check so idle accounts don't open a claim
// transaction every pass. A stale watermark only ever
// produces a harmless extra claim.
pending, err := w.deliveries.HasForUserSince(ctx, rec.UserID,
Cursor{CreatedAt: rec.WatermarkCreatedAt, ID: rec.WatermarkID})
if err != nil {
w.logger.Warn("channel pass: pending check failed", "user_id", rec.UserID, "error", err)
continue
}
if !pending {
continue
}
case ChannelModeDailyDigest:
if !digestDue {
continue
}
default:
continue
}
if err := w.processAccount(ctx, rec); err != nil {
if errors.Is(err, errChannelUnavailable) {
return // channel turned off mid-pass; nothing else will send either
}
failures++
w.logger.Warn("channel send failed", "user_id", rec.UserID, "mode", mode, "error", err)
}
}
}
// processAccount sends one account's pending notifications under the prefs
// row lock. The send happens inside the claim transaction: the row lock is
// per-account and only contends with other nodes, and committing the
// watermark only after a successful send is what makes the channel durable.
func (w *accountChannelWorker) processAccount(ctx context.Context, rec accountRecipient) error {
tx, err := w.pool.Begin(ctx)
if err != nil {
return fmt.Errorf("begin channel dispatch tx: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
claimed, err := w.channel.claim(ctx, tx, rec.UserID)
if err != nil {
return err
}
if claimed == nil {
return nil // another node is handling this account
}
// Re-derive eligibility from the locked row: the pre-scan snapshot may
// predate a user mode flip or another node's digest stamp.
mode := effectiveChannelMode(claimed.Mode, w.channel.allowPerEpisode(ctx))
digestDue := channelDigestDue(w.now(), w.channel.digestHour(ctx), claimed.LastDigestAt)
// sendKind is the rendering the channel applies (per-episode alert vs
// digest summary); for the combined mode it differs from the stored mode.
sendKind := mode
since := Cursor{CreatedAt: claimed.WatermarkCreatedAt, ID: claimed.WatermarkID}
// fetchFrom is where this send reads rows from. Per-episode legs read
// from the watermark (unsent rows); a combined-mode digest recaps the
// whole window since the previous digest, which is usually behind the
// watermark because its items already went out individually.
fetchFrom := since
var digestAt *time.Time
switch mode {
case ChannelModePerEpisode:
case ChannelModeDailyDigest:
if !digestDue {
return nil
}
now := w.now()
digestAt = &now
case ChannelModePerEpisodeAndDigest:
if digestDue {
sendKind = ChannelModeDailyDigest
now := w.now()
digestAt = &now
if claimed.LastDigestAt != nil {
digestCursor := Cursor{CreatedAt: *claimed.LastDigestAt}
if cursorLess(digestCursor, fetchFrom) {
fetchFrom = digestCursor
}
}
} else {
sendKind = ChannelModePerEpisode
}
default:
return nil
}
rows, err := w.deliveries.ListForUserSince(ctx, tx, rec.UserID, fetchFrom, channelFetchLimit)
if err != nil {
return err
}
if len(rows) == 0 {
// Nothing new. Digests still stamp so eligibility stops re-checking
// until tomorrow; the watermark needs no update.
if digestAt != nil {
if err := w.channel.markSent(ctx, tx, rec.UserID, since, digestAt); err != nil {
return err
}
return tx.Commit(ctx)
}
return nil
}
items := rows
if mode == ChannelModeDailyDigest {
// The digest-only mode reports what the user hasn't seen; rows
// already read in another client are skipped but the watermark still
// passes them. The combined mode's digest deliberately recaps
// everything — its per-episode sends already covered the new rows.
items = make([]DeliveryRow, 0, len(rows))
for _, row := range rows {
if row.ReadAt == nil {
items = append(items, row)
}
}
}
last := rows[len(rows)-1]
watermark := Cursor{CreatedAt: last.CreatedAt, ID: last.ID}
if cursorLess(watermark, since) {
// A combined-mode digest can read entirely behind the watermark;
// the watermark only ever moves forward.
watermark = since
}
if len(items) > 0 {
if err := w.channel.send(ctx, tx, rec.UserID, sendKind, items); err != nil {
if errors.Is(err, errChannelUnavailable) {
return err
}
if markErr := w.channel.markFailure(ctx, tx, rec.UserID, err); markErr != nil {
return errors.Join(err, markErr)
}
if commitErr := tx.Commit(ctx); commitErr != nil {
return errors.Join(err, commitErr)
}
return err
}
w.logger.Info("notification sent",
"user_id", rec.UserID, "mode", mode, "items", len(items))
}
if err := w.channel.markSent(ctx, tx, rec.UserID, watermark, digestAt); err != nil {
return err
}
return tx.Commit(ctx)
}
// nudgeDispatcher plugs an account-channel worker into the MultiDispatcher: a
// new delivery just nudges the sweep, which reads everything since the
// watermark. No per-delivery state is kept, so dropped nudges cost only poll
// latency.
type nudgeDispatcher struct {
worker *accountChannelWorker
}
func newNudgeDispatcher(worker *accountChannelWorker) *nudgeDispatcher {
return &nudgeDispatcher{worker: worker}
}
// Dispatch implements Dispatcher.
func (d *nudgeDispatcher) Dispatch(_ context.Context, _ DeliveryRow) error {
if d != nil {
d.worker.Nudge()
}
return nil
}
+18
View File
@@ -244,6 +244,24 @@ func (r *DeliveryRepository) ListForUserSince(ctx context.Context, tx pgx.Tx, us
return scanDeliveryRows(rows)
}
// HasForUserSince reports whether the account has any delivery newer than the
// given watermark. Cheap pre-check (index-only) so account-channel sweeps do
// not open a claim transaction for idle accounts every pass.
func (r *DeliveryRepository) HasForUserSince(ctx context.Context, userID int, since Cursor) (bool, error) {
var exists bool
err := r.pool.QueryRow(ctx, `
SELECT EXISTS (
SELECT 1 FROM notification_deliveries
WHERE user_id = $1 AND (created_at, id) > ($2, $3)
)`,
userID, since.CreatedAt, since.ID,
).Scan(&exists)
if err != nil {
return false, fmt.Errorf("check deliveries since watermark: %w", err)
}
return exists, nil
}
// RecentUnread returns the newest unread rows for the websocket snapshot.
func (r *DeliveryRepository) RecentUnread(ctx context.Context, profileID string, limit int) ([]DeliveryRow, error) {
rows, err := r.pool.Query(ctx,
+248
View File
@@ -0,0 +1,248 @@
package notifications
import (
"context"
"errors"
"fmt"
"time"
"github.com/Silo-Server/silo-server/internal/discord"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// discordDMBlockedMessage is the link_failure text surfaced in the settings
// UI when Discord refuses the DM (error 50007).
const discordDMBlockedMessage = "Discord rejected the direct message. " +
"Make sure you share a server with the bot and allow direct messages from server members."
// discordChannel implements accountChannel over the Discord bot REST API.
// The bot token is read live from settings on every pass, so admin changes
// apply without a restart (same pattern as the SMTP sender).
type discordChannel struct {
prefs *DiscordPrefsRepository
settings *Settings
client *discord.Client
}
// newDiscordWorker assembles the Discord DM channel on the shared
// account-channel engine.
func newDiscordWorker(
pool *pgxpool.Pool,
deliveries *DeliveryRepository,
prefs *DiscordPrefsRepository,
settings *Settings,
client *discord.Client,
) *accountChannelWorker {
return newAccountChannelWorker(pool, deliveries, &discordChannel{
prefs: prefs,
settings: settings,
client: client,
})
}
func (c *discordChannel) name() string { return "discord" }
func (c *discordChannel) enabled(ctx context.Context) bool {
return c.settings.DiscordEnabled(ctx) && c.settings.DiscordBotToken(ctx) != ""
}
func (c *discordChannel) allowPerEpisode(ctx context.Context) bool {
return c.settings.DiscordAllowPerEpisode(ctx)
}
func (c *discordChannel) digestHour(ctx context.Context) int {
return c.settings.DiscordDigestHour(ctx)
}
func (c *discordChannel) listRecipients(ctx context.Context) ([]accountRecipient, error) {
return c.prefs.ListActiveRecipients(ctx)
}
func (c *discordChannel) claim(ctx context.Context, tx pgx.Tx, userID int) (*accountRecipient, error) {
return c.prefs.claimForUpdate(ctx, tx, userID)
}
func (c *discordChannel) markSent(ctx context.Context, tx pgx.Tx, userID int, watermark Cursor, digestAt *time.Time) error {
return c.prefs.markSent(ctx, tx, userID, watermark, digestAt)
}
func (c *discordChannel) markFailure(ctx context.Context, tx pgx.Tx, userID int, sendErr error) error {
message := "Discord delivery failed"
switch {
case errors.Is(sendErr, discord.ErrDMBlocked):
message = discordDMBlockedMessage
case sendErr != nil:
message = truncateWithEllipsis("Discord delivery failed: "+sendErr.Error(), 300)
}
return c.prefs.markFailure(ctx, tx, userID, message)
}
// send delivers one account's pending rows as a single bot DM. Failures that
// indicate a global problem (missing/rejected bot token, rate limiting) wrap
// errChannelUnavailable so the pass aborts without penalizing the account;
// everything else (notably 50007 DM-blocked) backs off per account and is
// surfaced as link health.
func (c *discordChannel) send(ctx context.Context, tx pgx.Tx, userID int, _ string, rows []DeliveryRow) error {
botToken := c.settings.DiscordBotToken(ctx)
if botToken == "" {
return fmt.Errorf("bot token not configured: %w", errChannelUnavailable)
}
discordUserID, dmChannelID, err := c.prefs.identityForSend(ctx, tx, userID)
if err != nil {
return err
}
if discordUserID == "" {
return fmt.Errorf("account %d has no linked discord identity", userID)
}
if dmChannelID == "" {
dmChannelID, err = c.client.OpenDMChannel(ctx, botToken, discordUserID)
if err != nil {
return classifyDiscordSendError(err)
}
if err := c.prefs.cacheDMChannel(ctx, tx, userID, dmChannelID); err != nil {
return err
}
}
payload, err := BuildDiscordDMPayload(rows)
if err != nil {
return fmt.Errorf("build discord dm payload: %w", err)
}
if err := c.client.SendDM(ctx, botToken, dmChannelID, payload); err != nil {
return classifyDiscordSendError(err)
}
return nil
}
// classifyDiscordSendError separates global transport problems (which abort
// the pass) from per-account failures (which back off and surface as link
// health).
func classifyDiscordSendError(err error) error {
if errors.Is(err, discord.ErrUnauthorized) || errors.Is(err, discord.ErrRateLimited) {
return fmt.Errorf("%w: %w", err, errChannelUnavailable)
}
return err
}
// Errors surfaced by the Discord System methods for the API layer to map to
// 4xx responses.
var (
ErrDiscordModeInvalid = errors.New("invalid discord notification mode")
ErrDiscordModeNotAllowed = errors.New("per-episode discord DMs are disabled by the administrator")
ErrDiscordNotLinked = errors.New("no linked discord account")
ErrDiscordNotConfigured = errors.New("discord integration is not configured")
)
// discordLinkStateTTL bounds how long a started link flow stays redeemable.
const discordLinkStateTTL = 10 * time.Minute
// DiscordConfigured reports whether the admin has supplied the full Discord
// application credential set (linking needs client ID + secret; DM delivery
// needs the bot token).
func (s *System) DiscordConfigured(ctx context.Context) bool {
return s != nil && s.DiscordPrefs != nil &&
s.Settings.DiscordClientID(ctx) != "" &&
s.Settings.DiscordClientSecret(ctx) != "" &&
s.Settings.DiscordBotToken(ctx) != ""
}
// DiscordAvailable reports whether the Discord DM channel can deliver right
// now: fully configured and the kill switch is on.
func (s *System) DiscordAvailable(ctx context.Context) bool {
return s.DiscordConfigured(ctx) && s.Settings.DiscordEnabled(ctx)
}
// DiscordPrefsFor returns the account's Discord link + mode state.
func (s *System) DiscordPrefsFor(ctx context.Context, userID int) (DiscordPrefs, error) {
if s == nil || s.DiscordPrefs == nil {
return DiscordPrefs{UserID: userID, Mode: ChannelModeOff}, nil
}
return s.DiscordPrefs.Get(ctx, userID)
}
// SetDiscordMode validates and stores the account's Discord DM mode.
// Enabling requires a linked Discord account and, for per-episode, the admin
// allowance.
func (s *System) SetDiscordMode(ctx context.Context, userID int, mode string) error {
if s == nil || s.DiscordPrefs == nil {
return ErrDiscordModeInvalid
}
if !ValidChannelMode(mode) {
return ErrDiscordModeInvalid
}
if ModeIncludesPerEpisode(mode) && !s.Settings.DiscordAllowPerEpisode(ctx) {
return ErrDiscordModeNotAllowed
}
return s.DiscordPrefs.SetMode(ctx, userID, mode)
}
// BeginDiscordLink records a one-time state row for a link flow started by
// userID.
func (s *System) BeginDiscordLink(ctx context.Context, state string, userID int) error {
if s == nil || s.DiscordPrefs == nil {
return ErrDiscordNotConfigured
}
return s.DiscordPrefs.CreateLinkState(ctx, state, userID, time.Now().UTC().Add(discordLinkStateTTL))
}
// ConsumeDiscordLinkState redeems a one-time link state, returning the user
// who started the flow. ok is false for unknown, used, or expired states.
func (s *System) ConsumeDiscordLinkState(ctx context.Context, state string) (int, bool, error) {
if s == nil || s.DiscordPrefs == nil {
return 0, false, ErrDiscordNotConfigured
}
return s.DiscordPrefs.ConsumeLinkState(ctx, state)
}
// CompleteDiscordLink exchanges the OAuth authorization code, resolves the
// Discord identity behind it, and links it to the account.
func (s *System) CompleteDiscordLink(ctx context.Context, userID int, code, redirectURI string) (discord.User, error) {
if s == nil || s.discordClient == nil || s.DiscordPrefs == nil {
return discord.User{}, ErrDiscordNotConfigured
}
clientID := s.Settings.DiscordClientID(ctx)
clientSecret := s.Settings.DiscordClientSecret(ctx)
if clientID == "" || clientSecret == "" {
return discord.User{}, ErrDiscordNotConfigured
}
accessToken, err := s.discordClient.ExchangeCode(ctx, clientID, clientSecret, code, redirectURI)
if err != nil {
return discord.User{}, err
}
user, err := s.discordClient.GetUser(ctx, accessToken)
if err != nil {
return discord.User{}, err
}
if err := s.DiscordPrefs.SetIdentity(ctx, userID, user.ID, user.Username); err != nil {
return discord.User{}, err
}
s.logger.Info("discord account linked", "user_id", userID, "discord_user_id", user.ID)
return user, nil
}
// UnlinkDiscord removes the account's Discord identity and switches the
// channel off.
func (s *System) UnlinkDiscord(ctx context.Context, userID int) error {
if s == nil || s.DiscordPrefs == nil {
return nil
}
return s.DiscordPrefs.ClearIdentity(ctx, userID)
}
// TestDiscordBot verifies the configured bot token by fetching the bot's own
// identity. Used by the admin test endpoint.
func (s *System) TestDiscordBot(ctx context.Context) (discord.User, error) {
if s == nil || s.discordClient == nil {
return discord.User{}, ErrDiscordNotConfigured
}
// The admin typically tests seconds after saving; don't let the read
// cache report a stale "not configured".
s.Settings.Invalidate(SettingDiscordBotToken)
botToken := s.Settings.DiscordBotToken(ctx)
if botToken == "" {
return discord.User{}, ErrDiscordNotConfigured
}
return s.discordClient.GetBotUser(ctx, botToken)
}
@@ -0,0 +1,291 @@
package notifications
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// DiscordPrefs is one account's Discord DM notification state as the API
// surfaces it: the OAuth-linked identity, the user-chosen mode, and the last
// delivery failure (for link health in the settings UI).
type DiscordPrefs struct {
UserID int
DiscordUserID string
DiscordUsername string
Mode string
LinkFailure string
}
// Linked reports whether the account has completed the Discord OAuth link.
func (p DiscordPrefs) Linked() bool { return p.DiscordUserID != "" }
// DiscordPrefsRepository owns notification_discord_prefs and the one-time
// link-state rows for the OAuth flow.
type DiscordPrefsRepository struct {
pool *pgxpool.Pool
}
// NewDiscordPrefsRepository creates a DiscordPrefsRepository.
func NewDiscordPrefsRepository(pool *pgxpool.Pool) *DiscordPrefsRepository {
return &DiscordPrefsRepository{pool: pool}
}
// Get returns the account's Discord prefs; missing rows default to mode off
// and no linked identity.
func (r *DiscordPrefsRepository) Get(ctx context.Context, userID int) (DiscordPrefs, error) {
prefs := DiscordPrefs{UserID: userID, Mode: ChannelModeOff}
err := r.pool.QueryRow(ctx, `
SELECT discord_user_id, discord_username, mode, link_failure
FROM notification_discord_prefs WHERE user_id = $1`,
userID,
).Scan(&prefs.DiscordUserID, &prefs.DiscordUsername, &prefs.Mode, &prefs.LinkFailure)
if errors.Is(err, pgx.ErrNoRows) {
return prefs, nil
}
if err != nil {
return prefs, fmt.Errorf("get discord prefs: %w", err)
}
return prefs, nil
}
// SetIdentity stores the OAuth-linked Discord identity. The watermark resets
// to now so the backlog never floods a fresh link, the DM channel cache is
// cleared (the identity may have changed), and failure state is wiped so the
// first send happens promptly. An existing mode survives a re-link.
func (r *DiscordPrefsRepository) SetIdentity(ctx context.Context, userID int, discordUserID, discordUsername string) error {
_, err := r.pool.Exec(ctx, `
INSERT INTO notification_discord_prefs (user_id, discord_user_id, discord_username)
VALUES ($1, $2, $3)
ON CONFLICT (user_id) DO UPDATE SET
discord_user_id = EXCLUDED.discord_user_id,
discord_username = EXCLUDED.discord_username,
dm_channel_id = '',
watermark_created_at = now(),
watermark_id = '',
last_attempt_at = NULL,
consecutive_failures = 0,
link_failure = '',
updated_at = now()`,
userID, discordUserID, discordUsername)
if err != nil {
return fmt.Errorf("set discord identity: %w", err)
}
return nil
}
// ClearIdentity unlinks the Discord account and switches the channel off.
func (r *DiscordPrefsRepository) ClearIdentity(ctx context.Context, userID int) error {
_, err := r.pool.Exec(ctx, `
UPDATE notification_discord_prefs SET
discord_user_id = '',
discord_username = '',
dm_channel_id = '',
mode = 'off',
last_attempt_at = NULL,
consecutive_failures = 0,
link_failure = '',
updated_at = now()
WHERE user_id = $1`,
userID)
if err != nil {
return fmt.Errorf("clear discord identity: %w", err)
}
return nil
}
// SetMode updates the account's Discord mode. Enabling from off resets the
// watermark to now so the backlog never floods a fresh opt-in, and clears
// failure backoff so the first send happens promptly. Returns
// ErrDiscordNotLinked when no linked identity exists.
func (r *DiscordPrefsRepository) SetMode(ctx context.Context, userID int, mode string) error {
if !ValidChannelMode(mode) {
return fmt.Errorf("invalid discord mode %q", mode)
}
tag, err := r.pool.Exec(ctx, `
UPDATE notification_discord_prefs SET
mode = $2,
watermark_created_at = CASE
WHEN notification_discord_prefs.mode = 'off' THEN now()
ELSE notification_discord_prefs.watermark_created_at
END,
watermark_id = CASE
WHEN notification_discord_prefs.mode = 'off' THEN ''
ELSE notification_discord_prefs.watermark_id
END,
last_attempt_at = NULL,
consecutive_failures = 0,
updated_at = now()
WHERE user_id = $1 AND discord_user_id <> ''`,
userID, mode)
if err != nil {
return fmt.Errorf("set discord mode: %w", err)
}
if tag.RowsAffected() == 0 {
// Switching off without a linked row is already the desired state
// (e.g. a mode update racing an unlink); only enabling needs a link.
if mode == ChannelModeOff {
return nil
}
return ErrDiscordNotLinked
}
return nil
}
// ListActiveRecipients returns every linked account with Discord DMs on.
// Disabled or deleted accounts drop out of the join.
func (r *DiscordPrefsRepository) ListActiveRecipients(ctx context.Context) ([]accountRecipient, error) {
rows, err := r.pool.Query(ctx, `
SELECT p.user_id, p.mode, p.watermark_created_at, p.watermark_id,
p.last_digest_at, p.last_attempt_at, p.consecutive_failures
FROM notification_discord_prefs p
JOIN users u ON u.id = p.user_id AND u.enabled
WHERE p.mode <> 'off' AND p.discord_user_id <> ''
ORDER BY p.user_id`)
if err != nil {
return nil, fmt.Errorf("list discord recipients: %w", err)
}
defer rows.Close()
out := make([]accountRecipient, 0, 8)
for rows.Next() {
var rec accountRecipient
if err := rows.Scan(&rec.UserID, &rec.Mode, &rec.WatermarkCreatedAt, &rec.WatermarkID,
&rec.LastDigestAt, &rec.LastAttemptAt, &rec.ConsecutiveFailures); err != nil {
return nil, fmt.Errorf("scan discord recipient: %w", err)
}
out = append(out, rec)
}
return out, rows.Err()
}
// claimForUpdate locks the account's prefs row for one dispatch attempt.
// SKIP LOCKED makes concurrent nodes pass over each other's in-flight users
// instead of double-sending; (nil, nil) means another node holds the row.
func (r *DiscordPrefsRepository) claimForUpdate(ctx context.Context, tx pgx.Tx, userID int) (*accountRecipient, error) {
rec := accountRecipient{UserID: userID}
err := tx.QueryRow(ctx, `
SELECT mode, watermark_created_at, watermark_id, last_digest_at,
last_attempt_at, consecutive_failures
FROM notification_discord_prefs
WHERE user_id = $1 AND discord_user_id <> ''
FOR UPDATE SKIP LOCKED`,
userID,
).Scan(&rec.Mode, &rec.WatermarkCreatedAt, &rec.WatermarkID,
&rec.LastDigestAt, &rec.LastAttemptAt, &rec.ConsecutiveFailures)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("claim discord prefs: %w", err)
}
return &rec, nil
}
// identityForSend reads the locked row's identity fields inside the claim
// transaction, so the send targets exactly the identity the lock covers.
func (r *DiscordPrefsRepository) identityForSend(ctx context.Context, tx pgx.Tx, userID int) (discordUserID, dmChannelID string, err error) {
err = tx.QueryRow(ctx, `
SELECT discord_user_id, dm_channel_id
FROM notification_discord_prefs WHERE user_id = $1`,
userID,
).Scan(&discordUserID, &dmChannelID)
if err != nil {
return "", "", fmt.Errorf("read discord identity: %w", err)
}
return discordUserID, dmChannelID, nil
}
// cacheDMChannel stores a freshly opened DM channel ID under the claim lock.
func (r *DiscordPrefsRepository) cacheDMChannel(ctx context.Context, tx pgx.Tx, userID int, channelID string) error {
_, err := tx.Exec(ctx, `
UPDATE notification_discord_prefs SET dm_channel_id = $2, updated_at = now()
WHERE user_id = $1`,
userID, channelID)
if err != nil {
return fmt.Errorf("cache dm channel: %w", err)
}
return nil
}
// markSent advances the watermark past everything the DM covered and resets
// failure and link-health state. digestAt is non-nil for digest sends.
func (r *DiscordPrefsRepository) markSent(ctx context.Context, tx pgx.Tx, userID int, watermark Cursor, digestAt *time.Time) error {
_, err := tx.Exec(ctx, `
UPDATE notification_discord_prefs SET
watermark_created_at = $2,
watermark_id = $3,
last_digest_at = COALESCE($4, last_digest_at),
last_attempt_at = now(),
consecutive_failures = 0,
link_failure = '',
updated_at = now()
WHERE user_id = $1`,
userID, watermark.CreatedAt, watermark.ID, digestAt)
if err != nil {
return fmt.Errorf("mark discord sent: %w", err)
}
return nil
}
// markFailure records a failed send for backoff and surfaces the failure in
// the settings UI; the watermark stays put so the next eligible pass retries
// the same items.
func (r *DiscordPrefsRepository) markFailure(ctx context.Context, tx pgx.Tx, userID int, message string) error {
_, err := tx.Exec(ctx, `
UPDATE notification_discord_prefs SET
last_attempt_at = now(),
consecutive_failures = consecutive_failures + 1,
link_failure = $2,
updated_at = now()
WHERE user_id = $1`,
userID, message)
if err != nil {
return fmt.Errorf("mark discord failure: %w", err)
}
return nil
}
// CreateLinkState records a one-time OAuth state row for the link flow.
func (r *DiscordPrefsRepository) CreateLinkState(ctx context.Context, state string, userID int, expiresAt time.Time) error {
_, err := r.pool.Exec(ctx, `
INSERT INTO notification_discord_link_state (state, user_id, expires_at)
VALUES ($1, $2, $3)`,
state, userID, expiresAt)
if err != nil {
return fmt.Errorf("create discord link state: %w", err)
}
return nil
}
// ConsumeLinkState atomically deletes and returns an unexpired link-state
// row, making each state single-use. ok is false for unknown, already-used,
// or expired states.
func (r *DiscordPrefsRepository) ConsumeLinkState(ctx context.Context, state string) (userID int, ok bool, err error) {
err = r.pool.QueryRow(ctx, `
DELETE FROM notification_discord_link_state
WHERE state = $1 AND expires_at > now()
RETURNING user_id`,
state,
).Scan(&userID)
if errors.Is(err, pgx.ErrNoRows) {
return 0, false, nil
}
if err != nil {
return 0, false, fmt.Errorf("consume discord link state: %w", err)
}
return userID, true, nil
}
// DeleteExpiredLinkStates reaps abandoned link-flow rows (retention).
func (r *DiscordPrefsRepository) DeleteExpiredLinkStates(ctx context.Context) (int64, error) {
tag, err := r.pool.Exec(ctx,
`DELETE FROM notification_discord_link_state WHERE expires_at <= now()`)
if err != nil {
return 0, fmt.Errorf("delete expired discord link states: %w", err)
}
return tag.RowsAffected(), nil
}
+64 -275
View File
@@ -4,286 +4,94 @@ import (
"context"
"errors"
"fmt"
"log/slog"
"time"
"github.com/Silo-Server/silo-server/internal/mail"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
const (
emailPollInterval = time.Minute
// emailNudgeDelay coalesces the per-row dispatch nudges of one fanout
// batch (all rows commit before the first nudge fires) into one pass.
emailNudgeDelay = 2 * time.Second
// emailFetchLimit bounds one email's worth of watermark progress; the
// next pass drains the remainder.
emailFetchLimit = 200
// emailMaxFailuresPerPass stops a pass early when sends keep failing —
// SMTP trouble is almost always global, not per-recipient.
emailMaxFailuresPerPass = 3
emailFailureBackoffBase = time.Minute
emailFailureBackoffMax = 6 * time.Hour
)
// effectiveEmailMode coerces per-episode to the daily digest when the admin
// has disallowed per-episode email, instead of silencing those accounts.
func effectiveEmailMode(mode string, allowPerEpisode bool) string {
if mode == EmailModePerEpisode && !allowPerEpisode {
return EmailModeDailyDigest
}
return mode
}
// emailDigestDue reports whether a daily digest should go out: today's send
// time (digestHour, local) has passed and no digest was stamped since.
func emailDigestDue(now time.Time, digestHour int, lastDigestAt *time.Time) bool {
todaySend := time.Date(now.Year(), now.Month(), now.Day(), digestHour, 0, 0, 0, now.Location())
if now.Before(todaySend) {
return false
}
return lastDigestAt == nil || lastDigestAt.Before(todaySend)
}
// emailRetryEligible applies exponential backoff after failed sends:
// 1m, 2m, 4m, ... capped at emailFailureBackoffMax.
func emailRetryEligible(now time.Time, lastAttemptAt *time.Time, consecutiveFailures int) bool {
if consecutiveFailures <= 0 || lastAttemptAt == nil {
return true
}
backoff := emailFailureBackoffBase << min(consecutiveFailures-1, 30)
if backoff > emailFailureBackoffMax || backoff <= 0 {
backoff = emailFailureBackoffMax
}
return !now.Before(lastAttemptAt.Add(backoff))
}
// EmailWorker delivers notification emails. Unlike webhooks and web push it
// keeps no per-target outbox: deliveries already carry user_id, so a per-user
// watermark over notification_deliveries is the durable dispatch state. The
// watermark advances only after a successful SMTP send, and one email covers
// everything since the last one — which also collapses the duplicate rows an
// account gets when several of its profiles follow the same series.
type EmailWorker struct {
pool *pgxpool.Pool
deliveries *DeliveryRepository
prefs *EmailPrefsRepository
settings *Settings
sender mail.Sender
logger *slog.Logger
nudge chan struct{}
now func() time.Time
// emailChannel implements accountChannel over the shared SMTP core. The
// engine owns the sweep loop and watermark; this adapter only knows how to
// list/claim email prefs rows and compose+send one account's message.
type emailChannel struct {
prefs *EmailPrefsRepository
settings *Settings
sender mail.Sender
}
// newEmailWorker assembles the email channel on the shared account-channel
// engine.
func newEmailWorker(
pool *pgxpool.Pool,
deliveries *DeliveryRepository,
prefs *EmailPrefsRepository,
settings *Settings,
sender mail.Sender,
) *EmailWorker {
return &EmailWorker{
pool: pool,
deliveries: deliveries,
prefs: prefs,
settings: settings,
sender: sender,
logger: slog.Default().With("component", "notifications.email"),
nudge: make(chan struct{}, 1),
now: time.Now,
}
) *accountChannelWorker {
return newAccountChannelWorker(pool, deliveries, &emailChannel{
prefs: prefs,
settings: settings,
sender: sender,
})
}
// Nudge schedules a near-term pass so per-episode emails follow fanout within
// seconds instead of waiting for the next poll. Non-blocking.
func (w *EmailWorker) Nudge() {
if w == nil {
return
}
select {
case w.nudge <- struct{}{}:
default:
}
func (c *emailChannel) name() string { return "email" }
func (c *emailChannel) enabled(ctx context.Context) bool {
return c.settings.EmailEnabled(ctx) && c.sender.Enabled(ctx)
}
// Run sweeps eligible accounts until ctx is canceled.
func (w *EmailWorker) Run(ctx context.Context) {
ticker := time.NewTicker(emailPollInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
case <-w.nudge:
select {
case <-ctx.Done():
return
case <-time.After(emailNudgeDelay):
}
}
if !w.settings.EmailEnabled(ctx) || !w.sender.Enabled(ctx) {
continue
}
w.runPass(ctx)
}
func (c *emailChannel) allowPerEpisode(ctx context.Context) bool {
return c.settings.EmailAllowPerEpisode(ctx)
}
// runPass attempts one send per eligible account. Failures back off per
// account; the pass aborts entirely on ErrNotConfigured or after a few
// consecutive failures, since both indicate a global SMTP problem.
func (w *EmailWorker) runPass(ctx context.Context) {
recipients, err := w.prefs.ListActiveRecipients(ctx)
if err != nil {
w.logger.Error("email pass: list recipients failed", "error", err)
return
}
if len(recipients) == 0 {
return
}
allowPerEpisode := w.settings.EmailAllowPerEpisode(ctx)
digestHour := w.settings.EmailDigestHour(ctx)
now := w.now()
failures := 0
for _, rec := range recipients {
if ctx.Err() != nil || failures >= emailMaxFailuresPerPass {
return
}
if !emailRetryEligible(now, rec.LastAttemptAt, rec.ConsecutiveFailures) {
continue
}
mode := effectiveEmailMode(rec.Mode, allowPerEpisode)
switch mode {
case EmailModePerEpisode:
// Cheap pre-check so idle accounts don't open a claim
// transaction every pass. A stale watermark only ever
// produces a harmless extra claim.
pending, err := w.prefs.HasDeliveriesSince(ctx, rec.UserID,
Cursor{CreatedAt: rec.WatermarkCreatedAt, ID: rec.WatermarkID})
if err != nil {
w.logger.Warn("email pass: pending check failed", "user_id", rec.UserID, "error", err)
continue
}
if !pending {
continue
}
case EmailModeDailyDigest:
if !emailDigestDue(now, digestHour, rec.LastDigestAt) {
continue
}
default:
continue
}
if err := w.processAccount(ctx, rec); err != nil {
if errors.Is(err, mail.ErrNotConfigured) {
return // email turned off mid-pass; nothing else will send either
}
failures++
w.logger.Warn("email send failed", "user_id", rec.UserID, "mode", mode, "error", err)
}
}
func (c *emailChannel) digestHour(ctx context.Context) int {
return c.settings.EmailDigestHour(ctx)
}
// processAccount sends one account's pending notifications under the prefs
// row lock. The SMTP send happens inside the claim transaction: the row lock
// is per-account and only contends with other nodes, and committing the
// watermark only after a successful send is what makes the channel durable.
func (w *EmailWorker) processAccount(ctx context.Context, rec emailRecipient) error {
tx, err := w.pool.Begin(ctx)
func (c *emailChannel) listRecipients(ctx context.Context) ([]accountRecipient, error) {
return c.prefs.ListActiveRecipients(ctx)
}
func (c *emailChannel) claim(ctx context.Context, tx pgx.Tx, userID int) (*accountRecipient, error) {
return c.prefs.claimForUpdate(ctx, tx, userID)
}
func (c *emailChannel) markSent(ctx context.Context, tx pgx.Tx, userID int, watermark Cursor, digestAt *time.Time) error {
return c.prefs.markSent(ctx, tx, userID, watermark, digestAt)
}
func (c *emailChannel) markFailure(ctx context.Context, tx pgx.Tx, userID int, _ error) error {
return c.prefs.markFailure(ctx, tx, userID)
}
// send composes and sends one account's pending notifications. The address is
// re-read under the claim so a mid-pass address removal fails cleanly instead
// of sending to a stale recipient.
func (c *emailChannel) send(ctx context.Context, tx pgx.Tx, userID int, mode string, rows []DeliveryRow) error {
var email string
err := tx.QueryRow(ctx,
`SELECT COALESCE(email, '') FROM users WHERE id = $1 AND enabled`, userID,
).Scan(&email)
if errors.Is(err, pgx.ErrNoRows) || (err == nil && email == "") {
return fmt.Errorf("account %d has no usable email address", userID)
}
if err != nil {
return fmt.Errorf("begin email dispatch tx: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
claimed, err := w.prefs.claimForUpdate(ctx, tx, rec.UserID)
if err != nil {
return err
}
if claimed == nil {
return nil // another node is handling this account
return fmt.Errorf("look up account email: %w", err)
}
// Re-derive eligibility from the locked row: the pre-scan snapshot may
// predate a user mode flip or another node's digest stamp.
mode := effectiveEmailMode(claimed.Mode, w.settings.EmailAllowPerEpisode(ctx))
switch mode {
case EmailModePerEpisode:
case EmailModeDailyDigest:
if !emailDigestDue(w.now(), w.settings.EmailDigestHour(ctx), claimed.LastDigestAt) {
return nil
}
default:
return nil
content := composeNotificationEmail(mode, rows, c.settings.EmailExternalURL(ctx))
err = c.sender.Send(ctx, mail.Message{
To: []string{email},
Subject: content.Subject,
TextBody: content.Text,
HTMLBody: content.HTML,
})
if errors.Is(err, mail.ErrNotConfigured) {
return fmt.Errorf("smtp not configured: %w", errChannelUnavailable)
}
since := Cursor{CreatedAt: claimed.WatermarkCreatedAt, ID: claimed.WatermarkID}
rows, err := w.deliveries.ListForUserSince(ctx, tx, rec.UserID, since, emailFetchLimit)
if err != nil {
return err
}
var digestAt *time.Time
if mode == EmailModeDailyDigest {
now := w.now()
digestAt = &now
}
if len(rows) == 0 {
// Nothing new. Digests still stamp so eligibility stops re-checking
// until tomorrow; the watermark needs no update.
if digestAt != nil {
if err := w.prefs.markSent(ctx, tx, rec.UserID, since, digestAt); err != nil {
return err
}
return tx.Commit(ctx)
}
return nil
}
items := rows
if mode == EmailModeDailyDigest {
// The digest reports what the user hasn't seen; rows already read in
// another client are skipped but the watermark still passes them.
items = make([]DeliveryRow, 0, len(rows))
for _, row := range rows {
if row.ReadAt == nil {
items = append(items, row)
}
}
}
last := rows[len(rows)-1]
watermark := Cursor{CreatedAt: last.CreatedAt, ID: last.ID}
if len(items) > 0 {
content := composeNotificationEmail(mode, items, w.settings.EmailExternalURL(ctx))
err = w.sender.Send(ctx, mail.Message{
To: []string{rec.Email},
Subject: content.Subject,
TextBody: content.Text,
HTMLBody: content.HTML,
})
if err != nil {
if errors.Is(err, mail.ErrNotConfigured) {
return err
}
if markErr := w.prefs.markFailure(ctx, tx, rec.UserID); markErr != nil {
return errors.Join(err, markErr)
}
if commitErr := tx.Commit(ctx); commitErr != nil {
return errors.Join(err, commitErr)
}
return err
}
w.logger.Info("notification email sent",
"user_id", rec.UserID, "mode", mode, "items", len(items))
}
if err := w.prefs.markSent(ctx, tx, rec.UserID, watermark, digestAt); err != nil {
return err
}
return tx.Commit(ctx)
return err
}
// Errors surfaced by SetEmailMode for the API layer to map to 4xx responses.
@@ -319,10 +127,10 @@ func (s *System) SetEmailMode(ctx context.Context, userID int, mode string) erro
if s == nil || s.EmailPrefs == nil {
return ErrEmailModeInvalid
}
if !ValidEmailMode(mode) {
if !ValidChannelMode(mode) {
return ErrEmailModeInvalid
}
if mode == EmailModePerEpisode && !s.Settings.EmailAllowPerEpisode(ctx) {
if ModeIncludesPerEpisode(mode) && !s.Settings.EmailAllowPerEpisode(ctx) {
return ErrEmailModeNotAllowed
}
if mode != EmailModeOff {
@@ -339,22 +147,3 @@ func (s *System) SetEmailMode(ctx context.Context, userID int, mode string) erro
}
return s.EmailPrefs.SetMode(ctx, userID, mode)
}
// EmailDispatcher plugs the email worker into the MultiDispatcher: a new
// delivery just nudges the sweep, which reads everything since the watermark.
// No per-delivery state is kept, so dropped nudges cost only poll latency.
type EmailDispatcher struct {
worker *EmailWorker
}
func newEmailDispatcher(worker *EmailWorker) *EmailDispatcher {
return &EmailDispatcher{worker: worker}
}
// Dispatch implements Dispatcher.
func (d *EmailDispatcher) Dispatch(_ context.Context, _ DeliveryRow) error {
if d != nil {
d.worker.Nudge()
}
return nil
}
+23 -17
View File
@@ -7,65 +7,71 @@ import (
"time"
)
func TestEffectiveEmailMode(t *testing.T) {
if got := effectiveEmailMode(EmailModePerEpisode, true); got != EmailModePerEpisode {
func TestEffectiveChannelMode(t *testing.T) {
if got := effectiveChannelMode(ChannelModePerEpisode, true); got != ChannelModePerEpisode {
t.Fatalf("allowed per-episode coerced to %q", got)
}
if got := effectiveEmailMode(EmailModePerEpisode, false); got != EmailModeDailyDigest {
if got := effectiveChannelMode(ChannelModePerEpisode, false); got != ChannelModeDailyDigest {
t.Fatalf("disallowed per-episode should coerce to digest, got %q", got)
}
if got := effectiveEmailMode(EmailModeDailyDigest, false); got != EmailModeDailyDigest {
if got := effectiveChannelMode(ChannelModeDailyDigest, false); got != ChannelModeDailyDigest {
t.Fatalf("digest mode changed to %q", got)
}
if got := effectiveEmailMode(EmailModeOff, true); got != EmailModeOff {
if got := effectiveChannelMode(ChannelModeOff, true); got != ChannelModeOff {
t.Fatalf("off mode changed to %q", got)
}
if got := effectiveChannelMode(ChannelModePerEpisodeAndDigest, true); got != ChannelModePerEpisodeAndDigest {
t.Fatalf("allowed combined mode coerced to %q", got)
}
if got := effectiveChannelMode(ChannelModePerEpisodeAndDigest, false); got != ChannelModeDailyDigest {
t.Fatalf("disallowed combined mode should coerce to digest, got %q", got)
}
}
func TestEmailDigestDue(t *testing.T) {
func TestChannelDigestDue(t *testing.T) {
loc := time.UTC
morning := time.Date(2026, 6, 11, 7, 30, 0, 0, loc)
afternoon := time.Date(2026, 6, 11, 14, 0, 0, 0, loc)
yesterday := time.Date(2026, 6, 10, 9, 0, 0, 0, loc)
today := time.Date(2026, 6, 11, 8, 5, 0, 0, loc)
if emailDigestDue(morning, 8, nil) {
if channelDigestDue(morning, 8, nil) {
t.Fatal("digest due before today's send hour")
}
if !emailDigestDue(afternoon, 8, nil) {
if !channelDigestDue(afternoon, 8, nil) {
t.Fatal("first-ever digest not due after send hour")
}
if !emailDigestDue(afternoon, 8, &yesterday) {
if !channelDigestDue(afternoon, 8, &yesterday) {
t.Fatal("digest not due when last one was yesterday")
}
if emailDigestDue(afternoon, 8, &today) {
if channelDigestDue(afternoon, 8, &today) {
t.Fatal("digest due twice in one day")
}
}
func TestEmailRetryEligible(t *testing.T) {
func TestChannelRetryEligible(t *testing.T) {
now := time.Date(2026, 6, 11, 12, 0, 0, 0, time.UTC)
recent := now.Add(-30 * time.Second)
stale := now.Add(-10 * time.Minute)
if !emailRetryEligible(now, nil, 0) {
if !channelRetryEligible(now, nil, 0) {
t.Fatal("clean account not eligible")
}
if !emailRetryEligible(now, &recent, 0) {
if !channelRetryEligible(now, &recent, 0) {
t.Fatal("successful account not eligible")
}
if emailRetryEligible(now, &recent, 1) {
if channelRetryEligible(now, &recent, 1) {
t.Fatal("eligible 30s after first failure (backoff is 1m)")
}
if !emailRetryEligible(now, &stale, 3) {
if !channelRetryEligible(now, &stale, 3) {
t.Fatal("not eligible 10m after third failure (backoff is 4m)")
}
// Large failure counts must not overflow the shift; cap applies.
old := now.Add(-7 * time.Hour)
if !emailRetryEligible(now, &old, 60) {
if !channelRetryEligible(now, &old, 60) {
t.Fatal("not eligible past the 6h backoff cap")
}
if emailRetryEligible(now, &recent, 60) {
if channelRetryEligible(now, &recent, 60) {
t.Fatal("eligible 30s after many failures")
}
}
+16 -49
View File
@@ -12,22 +12,14 @@ import (
// Email notification modes. The channel is account-level: email addresses
// live on users, not profiles, so one setting covers every profile on the
// account and the worker collapses cross-profile duplicates.
// account and the worker collapses cross-profile duplicates. The values are
// the shared account-channel modes.
const (
EmailModeOff = "off"
EmailModePerEpisode = "per_episode"
EmailModeDailyDigest = "daily_digest"
EmailModeOff = ChannelModeOff
EmailModePerEpisode = ChannelModePerEpisode
EmailModeDailyDigest = ChannelModeDailyDigest
)
// ValidEmailMode reports whether mode is a recognized email mode value.
func ValidEmailMode(mode string) bool {
switch mode {
case EmailModeOff, EmailModePerEpisode, EmailModeDailyDigest:
return true
}
return false
}
// EmailPrefs is one account's email notification state: the user-chosen mode
// plus the worker's dispatch watermark and failure backoff counters.
type EmailPrefs struct {
@@ -40,12 +32,6 @@ type EmailPrefs struct {
ConsecutiveFailures int
}
// emailRecipient pairs an active prefs row with the account's email address.
type emailRecipient struct {
EmailPrefs
Email string
}
// EmailPrefsRepository owns notification_email_prefs.
type EmailPrefsRepository struct {
pool *pgxpool.Pool
@@ -79,7 +65,7 @@ func (r *EmailPrefsRepository) Get(ctx context.Context, userID int) (EmailPrefs,
// the row) resets the watermark to now so the backlog never floods a fresh
// opt-in, and clears failure backoff so the first send happens promptly.
func (r *EmailPrefsRepository) SetMode(ctx context.Context, userID int, mode string) error {
if !ValidEmailMode(mode) {
if !ValidChannelMode(mode) {
return fmt.Errorf("invalid email mode %q", mode)
}
_, err := r.pool.Exec(ctx, `
@@ -107,11 +93,10 @@ func (r *EmailPrefsRepository) SetMode(ctx context.Context, userID int, mode str
// ListActiveRecipients returns every account with email notifications on and
// a usable address. Disabled or deleted accounts drop out of the join.
func (r *EmailPrefsRepository) ListActiveRecipients(ctx context.Context) ([]emailRecipient, error) {
func (r *EmailPrefsRepository) ListActiveRecipients(ctx context.Context) ([]accountRecipient, error) {
rows, err := r.pool.Query(ctx, `
SELECT p.user_id, p.mode, p.watermark_created_at, p.watermark_id,
p.last_digest_at, p.last_attempt_at, p.consecutive_failures,
u.email
p.last_digest_at, p.last_attempt_at, p.consecutive_failures
FROM notification_email_prefs p
JOIN users u ON u.id = p.user_id AND u.enabled AND COALESCE(u.email, '') <> ''
WHERE p.mode <> 'off'
@@ -120,11 +105,11 @@ func (r *EmailPrefsRepository) ListActiveRecipients(ctx context.Context) ([]emai
return nil, fmt.Errorf("list email recipients: %w", err)
}
defer rows.Close()
out := make([]emailRecipient, 0, 8)
out := make([]accountRecipient, 0, 8)
for rows.Next() {
var rec emailRecipient
var rec accountRecipient
if err := rows.Scan(&rec.UserID, &rec.Mode, &rec.WatermarkCreatedAt, &rec.WatermarkID,
&rec.LastDigestAt, &rec.LastAttemptAt, &rec.ConsecutiveFailures, &rec.Email); err != nil {
&rec.LastDigestAt, &rec.LastAttemptAt, &rec.ConsecutiveFailures); err != nil {
return nil, fmt.Errorf("scan email recipient: %w", err)
}
out = append(out, rec)
@@ -135,8 +120,8 @@ func (r *EmailPrefsRepository) ListActiveRecipients(ctx context.Context) ([]emai
// claimForUpdate locks the account's prefs row for one dispatch attempt.
// SKIP LOCKED makes concurrent nodes pass over each other's in-flight users
// instead of double-sending; (nil, nil) means another node holds the row.
func (r *EmailPrefsRepository) claimForUpdate(ctx context.Context, tx pgx.Tx, userID int) (*EmailPrefs, error) {
prefs := EmailPrefs{UserID: userID}
func (r *EmailPrefsRepository) claimForUpdate(ctx context.Context, tx pgx.Tx, userID int) (*accountRecipient, error) {
rec := accountRecipient{UserID: userID}
err := tx.QueryRow(ctx, `
SELECT mode, watermark_created_at, watermark_id, last_digest_at,
last_attempt_at, consecutive_failures
@@ -144,15 +129,15 @@ func (r *EmailPrefsRepository) claimForUpdate(ctx context.Context, tx pgx.Tx, us
WHERE user_id = $1
FOR UPDATE SKIP LOCKED`,
userID,
).Scan(&prefs.Mode, &prefs.WatermarkCreatedAt, &prefs.WatermarkID,
&prefs.LastDigestAt, &prefs.LastAttemptAt, &prefs.ConsecutiveFailures)
).Scan(&rec.Mode, &rec.WatermarkCreatedAt, &rec.WatermarkID,
&rec.LastDigestAt, &rec.LastAttemptAt, &rec.ConsecutiveFailures)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("claim email prefs: %w", err)
}
return &prefs, nil
return &rec, nil
}
// markSent advances the watermark past everything the email covered and
@@ -190,21 +175,3 @@ func (r *EmailPrefsRepository) markFailure(ctx context.Context, tx pgx.Tx, userI
}
return nil
}
// HasDeliveriesSince reports whether the account has any delivery newer than
// the watermark. Cheap pre-check (index-only) so the per-episode sweep does
// not open a claim transaction for idle accounts every pass.
func (r *EmailPrefsRepository) HasDeliveriesSince(ctx context.Context, userID int, since Cursor) (bool, error) {
var exists bool
err := r.pool.QueryRow(ctx, `
SELECT EXISTS (
SELECT 1 FROM notification_deliveries
WHERE user_id = $1 AND (created_at, id) > ($2, $3)
)`,
userID, since.CreatedAt, since.ID,
).Scan(&exists)
if err != nil {
return false, fmt.Errorf("check deliveries since watermark: %w", err)
}
return exists, nil
}
+70 -4
View File
@@ -18,8 +18,9 @@ type SettingReader interface {
// Server-setting keys for the notification system. All keys are live (no
// restart required): consumers read them through Settings, which caches reads
// briefly. The enabled flags default to on and act as kill switches — except
// webhooks, which are opt-in and stay off until an admin enables them. Flood
// safety comes from per-library seed markers, not from staged flag flips.
// webhooks and Discord, which are opt-in and stay off until an admin enables
// them. Flood safety comes from per-library seed markers, not from staged
// flag flips.
const (
SettingReleaseEventsEnabled = "notifications.release_events_enabled"
SettingFanoutEnabled = "notifications.fanout_enabled"
@@ -40,6 +41,17 @@ const (
SettingEmailAllowPerEpisode = "notifications.email.allow_per_episode"
SettingEmailDigestHour = "notifications.email.digest_hour"
SettingEmailExternalURL = "notifications.email.external_url"
SettingDiscordEnabled = "notifications.discord_enabled"
SettingDiscordAllowPerEpisode = "notifications.discord.allow_per_episode"
SettingDiscordDigestHour = "notifications.discord.digest_hour"
// Discord application credentials live under the discord.* namespace
// (admin-configured, alongside email.smtp_*). The secret and bot token
// are registered in catalog.SensitiveSettingKeys and encrypted at rest.
SettingDiscordClientID = "discord.client_id"
SettingDiscordClientSecret = "discord.client_secret"
SettingDiscordBotToken = "discord.bot_token"
)
const (
@@ -49,7 +61,9 @@ const (
defaultRetentionReadDays = 90
defaultRetentionUnread = 180
defaultRetentionEventDays = 30
defaultEmailDigestHour = 8
// defaultDigestHour applies to every account-level digest channel
// (email, Discord).
defaultDigestHour = 8
settingsCacheTTL = 15 * time.Second
)
@@ -108,6 +122,20 @@ func (s *Settings) raw(ctx context.Context, key string) string {
return value
}
// Invalidate drops cached values so the next read hits the store. Admin test
// paths use it: a test typically runs seconds after a settings save, inside
// the read-cache TTL, and must see the just-saved value.
func (s *Settings) Invalidate(keys ...string) {
if s == nil {
return
}
s.mu.Lock()
for _, key := range keys {
delete(s.cache, key)
}
s.mu.Unlock()
}
func (s *Settings) boolSetting(ctx context.Context, key string, fallback bool) bool {
raw := strings.TrimSpace(strings.ToLower(s.raw(ctx, key)))
switch raw {
@@ -224,7 +252,7 @@ func (s *Settings) EmailAllowPerEpisode(ctx context.Context) bool {
// EmailDigestHour is the hour of day (0-23, server-local time) at which daily
// digest emails go out.
func (s *Settings) EmailDigestHour(ctx context.Context) int {
return s.intSetting(ctx, SettingEmailDigestHour, defaultEmailDigestHour, 0, 23)
return s.intSetting(ctx, SettingEmailDigestHour, defaultDigestHour, 0, 23)
}
// EmailExternalURL is the externally reachable base URL of this server, used
@@ -233,3 +261,41 @@ func (s *Settings) EmailDigestHour(ctx context.Context) int {
func (s *Settings) EmailExternalURL(ctx context.Context) string {
return strings.TrimRight(strings.TrimSpace(s.raw(ctx, SettingEmailExternalURL)), "/")
}
// DiscordEnabled is the master switch for the Discord bot integration. Like
// webhooks it is opt-in: while off, the channel never delivers, linking is
// refused, and clients are told the channel is unavailable (which hides the
// Discord section in user settings). Actual availability additionally
// requires the configured bot credentials.
func (s *Settings) DiscordEnabled(ctx context.Context) bool {
return s.boolSetting(ctx, SettingDiscordEnabled, false)
}
// DiscordAllowPerEpisode controls whether users may choose per-episode
// Discord DMs. When off, accounts set to per-episode are coerced to the daily
// digest instead of going silent.
func (s *Settings) DiscordAllowPerEpisode(ctx context.Context) bool {
return s.boolSetting(ctx, SettingDiscordAllowPerEpisode, true)
}
// DiscordDigestHour is the hour of day (0-23, server-local time) at which
// daily digest DMs go out.
func (s *Settings) DiscordDigestHour(ctx context.Context) int {
return s.intSetting(ctx, SettingDiscordDigestHour, defaultDigestHour, 0, 23)
}
// DiscordClientID is the Discord application's OAuth2 client ID.
func (s *Settings) DiscordClientID(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingDiscordClientID))
}
// DiscordClientSecret is the Discord application's OAuth2 client secret.
func (s *Settings) DiscordClientSecret(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingDiscordClientSecret))
}
// DiscordBotToken is the Discord bot token used to open DM channels and send
// messages.
func (s *Settings) DiscordBotToken(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingDiscordBotToken))
}
+42 -10
View File
@@ -9,6 +9,7 @@ import (
"sync"
"time"
"github.com/Silo-Server/silo-server/internal/discord"
evt "github.com/Silo-Server/silo-server/internal/events"
"github.com/Silo-Server/silo-server/internal/mail"
"github.com/Silo-Server/silo-server/internal/models"
@@ -52,9 +53,14 @@ type System struct {
WebPush *WebPushService
// EmailPrefs is nil when no mail sender was provided.
EmailPrefs *EmailPrefsRepository
// DiscordPrefs holds Discord DM link + mode state; the channel only
// delivers once an admin configures bot credentials in settings.
DiscordPrefs *DiscordPrefsRepository
mailSender mail.Sender
emailWorker *EmailWorker
mailSender mail.Sender
emailWorker *accountChannelWorker
discordWorker *accountChannelWorker
discordClient *discord.Client
webhookRepo *WebhookRepository
webhookDispatcher *WebhookDispatcher
@@ -131,13 +137,21 @@ func NewSystem(
// Email rides the shared SMTP core. Unlike the per-target channels it
// keeps no outbox: its dispatcher only nudges the watermark sweep.
var emailPrefs *EmailPrefsRepository
var emailWorker *EmailWorker
var emailWorker *accountChannelWorker
if mailSender != nil {
emailPrefs = NewEmailPrefsRepository(pool)
emailWorker = newEmailWorker(pool, deliveries, emailPrefs, settings, mailSender)
dispatchers = append(dispatchers, newEmailDispatcher(emailWorker))
dispatchers = append(dispatchers, newNudgeDispatcher(emailWorker))
}
// Discord DMs ride the same account-watermark engine as email. The
// channel is always wired (its credentials live in settings and may be
// configured at runtime); enabled() gates each pass on the bot token.
discordPrefs := NewDiscordPrefsRepository(pool)
discordClient := discord.NewClient()
discordWorker := newDiscordWorker(pool, deliveries, discordPrefs, settings, discordClient)
dispatchers = append(dispatchers, newNudgeDispatcher(discordWorker))
multiDispatcher := NewMultiDispatcher(dispatchers...)
fanout := NewFanoutWorker(pool, releases, interests, deliveries, preferences, settings, multiDispatcher)
if webhookRepo != nil {
@@ -163,8 +177,11 @@ func NewSystem(
Webhooks: webhookService,
WebPush: webPushService,
EmailPrefs: emailPrefs,
DiscordPrefs: discordPrefs,
mailSender: mailSender,
emailWorker: emailWorker,
discordWorker: discordWorker,
discordClient: discordClient,
webhookRepo: webhookRepo,
webhookDispatcher: webhookDispatcher,
webhookRetry: webhookRetry,
@@ -246,6 +263,13 @@ func (s *System) Start(ctx context.Context) {
s.emailWorker.Run(ctx)
}()
}
if s.discordWorker != nil {
s.wg.Add(1)
go func() {
defer s.wg.Done()
s.discordWorker.Run(ctx)
}()
}
if s.webPushDispatcher != nil {
s.wg.Add(1)
go func() {
@@ -577,12 +601,13 @@ func (s *System) batchResolveSeries(ctx context.Context, itemIDs map[string]stru
// RetentionStats reports what a retention pass removed.
type RetentionStats struct {
DeliveriesDeleted int64
EventsDeleted int64
StaleEventsDeleted int64
InterestPruned int64
WebhookAttemptsDeleted int64
WebPushAttemptsDeleted int64
DeliveriesDeleted int64
EventsDeleted int64
StaleEventsDeleted int64
InterestPruned int64
WebhookAttemptsDeleted int64
WebPushAttemptsDeleted int64
DiscordLinkStatesDeleted int64
}
// RunRetention applies the retention policy: read deliveries past the read
@@ -637,6 +662,13 @@ func (s *System) RunRetention(ctx context.Context) (RetentionStats, error) {
}
stats.WebPushAttemptsDeleted = attempts
}
if s.DiscordPrefs != nil {
states, err := s.DiscordPrefs.DeleteExpiredLinkStates(ctx)
if err != nil {
return stats, fmt.Errorf("prune discord link states: %w", err)
}
stats.DiscordLinkStatesDeleted = states
}
return stats, nil
}
@@ -53,10 +53,50 @@ type discordWebhookBody struct {
Username string `json:"username"`
}
// BuildDiscordWebhookPayload renders a delivery as a Discord embed. Pure
// function; enforces Discord's embed limits with the spec's truncation
// BuildDiscordWebhookPayload renders a delivery as a Discord webhook body.
// Pure function; enforces Discord's embed limits with the spec's truncation
// policy (description first, then drop fields right-to-left).
func BuildDiscordWebhookPayload(row DeliveryRow, test bool) ([]byte, error) {
return json.Marshal(discordWebhookBody{
Embeds: []discordEmbed{buildDiscordEmbed(row, test)},
Username: "Silo",
})
}
// discordDMMaxEmbeds is Discord's per-message embed cap.
const discordDMMaxEmbeds = 10
// discordDMBody is a bot channel-message body. Unlike webhook bodies it has
// no username override: a bot message always carries the bot's own identity.
type discordDMBody struct {
Content string `json:"content,omitempty"`
Embeds []discordEmbed `json:"embeds"`
}
// BuildDiscordDMPayload renders one account's pending deliveries as a single
// bot DM, one embed per item up to Discord's 10-embed cap. Overflow keeps the
// newest items and points at the Silo inbox for the rest, mirroring the email
// digest's rendering cap.
func BuildDiscordDMPayload(rows []DeliveryRow) ([]byte, error) {
overflow := 0
if len(rows) > discordDMMaxEmbeds {
overflow = len(rows) - discordDMMaxEmbeds
rows = rows[len(rows)-discordDMMaxEmbeds:]
}
embeds := make([]discordEmbed, 0, len(rows))
for _, row := range rows {
embeds = append(embeds, buildDiscordEmbed(row, false))
}
body := discordDMBody{Embeds: embeds}
if overflow > 0 {
body.Content = fmt.Sprintf("…and %d more in your Silo inbox", overflow)
}
return json.Marshal(body)
}
// buildDiscordEmbed renders one delivery as a Discord embed within all of
// Discord's per-embed limits.
func buildDiscordEmbed(row DeliveryRow, test bool) discordEmbed {
flags := parseReasonFlags(row.ReasonFlags)
title := discordEmbedTitle(row)
@@ -116,11 +156,7 @@ func BuildDiscordWebhookPayload(row DeliveryRow, test bool) ([]byte, error) {
embed.Timestamp = row.CreatedAt.UTC().Format(time.RFC3339)
}
enforceDiscordTotalLimit(&embed)
return json.Marshal(discordWebhookBody{
Embeds: []discordEmbed{embed},
Username: "Silo",
})
return embed
}
func discordEmbedTitle(row DeliveryRow) string {
@@ -0,0 +1,56 @@
-- +goose Up
-- +goose StatementBegin
-- Discord DM notification channel (docs/superpowers/plans/notifications).
-- Account-level like email: the OAuth-linked Discord identity belongs to the
-- login account (users), not a profile, so one row carries the identity, the
-- user-chosen mode, and the watermark sweep state. The worker advances the
-- watermark only after a successful DM, so a crash or Discord outage re-sends
-- instead of dropping; the watermark resets to now() on link and on enabling
-- so history never floods a fresh opt-in.
--
-- Deliberately no FK to users: notification tables stay FK-free toward
-- account/profile storage (see 20260611100000). Rows for deleted accounts
-- drop out of the recipient join and are inert.
CREATE TABLE public.notification_discord_prefs (
user_id integer PRIMARY KEY,
-- OAuth-linked identity; empty until the user completes the link flow.
discord_user_id text NOT NULL DEFAULT '',
discord_username text NOT NULL DEFAULT '',
-- DM channel cache from POST /users/@me/channels (DM channels are
-- permanent, so the cache never expires; it is only a saved round-trip).
dm_channel_id text NOT NULL DEFAULT '',
mode text NOT NULL DEFAULT 'off'
CHECK (mode IN ('off', 'per_episode', 'daily_digest')),
watermark_created_at timestamptz NOT NULL DEFAULT now(),
watermark_id text NOT NULL DEFAULT '',
last_digest_at timestamptz,
last_attempt_at timestamptz,
consecutive_failures integer NOT NULL DEFAULT 0,
-- Last DM failure surfaced in the settings UI ('' = healthy). Typically
-- Discord error 50007: the user does not share a server with the bot.
link_failure text NOT NULL DEFAULT '',
updated_at timestamptz NOT NULL DEFAULT now()
);
-- One-time state rows for the Discord OAuth linking flow. The state is a
-- 32-byte random value; the atomic DELETE RETURNING on callback makes each
-- state single-use, which is the flow's authentication. Expired rows are
-- reaped by retention.
CREATE TABLE public.notification_discord_link_state (
state text PRIMARY KEY,
user_id integer NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
expires_at timestamptz NOT NULL
);
CREATE INDEX notification_discord_link_state_expires_idx
ON public.notification_discord_link_state (expires_at);
-- The Discord sweep reads deliveries by account via
-- notification_deliveries_user_created_idx (created in 20260611201720).
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
DROP TABLE IF EXISTS public.notification_discord_link_state;
DROP TABLE IF EXISTS public.notification_discord_prefs;
-- +goose StatementEnd
@@ -0,0 +1,39 @@
-- +goose Up
-- +goose StatementBegin
-- Adds the combined account-channel mode: per-episode sends all day plus a
-- daily digest recapping everything since the previous digest. Applies to
-- both account-level channels (email, Discord DMs).
ALTER TABLE public.notification_email_prefs
DROP CONSTRAINT notification_email_prefs_mode_check;
ALTER TABLE public.notification_email_prefs
ADD CONSTRAINT notification_email_prefs_mode_check
CHECK (mode IN ('off', 'per_episode', 'daily_digest', 'per_episode_and_digest'));
ALTER TABLE public.notification_discord_prefs
DROP CONSTRAINT notification_discord_prefs_mode_check;
ALTER TABLE public.notification_discord_prefs
ADD CONSTRAINT notification_discord_prefs_mode_check
CHECK (mode IN ('off', 'per_episode', 'daily_digest', 'per_episode_and_digest'));
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
-- Coerce combined-mode rows to per_episode before restoring the narrower
-- constraint, so the rollback cannot fail on existing data.
UPDATE public.notification_email_prefs
SET mode = 'per_episode' WHERE mode = 'per_episode_and_digest';
UPDATE public.notification_discord_prefs
SET mode = 'per_episode' WHERE mode = 'per_episode_and_digest';
ALTER TABLE public.notification_email_prefs
DROP CONSTRAINT notification_email_prefs_mode_check;
ALTER TABLE public.notification_email_prefs
ADD CONSTRAINT notification_email_prefs_mode_check
CHECK (mode IN ('off', 'per_episode', 'daily_digest'));
ALTER TABLE public.notification_discord_prefs
DROP CONSTRAINT notification_discord_prefs_mode_check;
ALTER TABLE public.notification_discord_prefs
ADD CONSTRAINT notification_discord_prefs_mode_check
CHECK (mode IN ('off', 'per_episode', 'daily_digest'));
-- +goose StatementEnd
+29 -2
View File
@@ -2327,22 +2327,49 @@ export interface NotificationWebhookTestResult {
message?: string;
}
/** An account-level digest channel (email, Discord DMs). */
export interface NotificationAccountChannelCapability {
available: boolean;
modes: string[];
digest_hour: number;
}
export interface NotificationCapability {
in_app: { enabled: boolean };
apple_push: { available: boolean; provider: string; supported_modes: string[] };
android_push: { available: boolean; provider: string; supported_modes: string[] };
web_push: { available: boolean; public_key?: string };
webhooks: { available: boolean; max_per_profile: number; supported_types: string[] };
email: { available: boolean; modes: string[]; digest_hour: number };
email: NotificationAccountChannelCapability;
discord: NotificationAccountChannelCapability;
}
export type NotificationEmailMode = "off" | "per_episode" | "daily_digest";
export type NotificationChannelMode =
| "off"
| "per_episode"
| "daily_digest"
| "per_episode_and_digest";
export type NotificationEmailMode = NotificationChannelMode;
export type NotificationDiscordMode = NotificationChannelMode;
/** Account-level (not per-profile): one mode covers all profiles. */
export interface NotificationEmailPreferences {
mode: NotificationEmailMode;
}
/** Account-level Discord DM channel: link state, mode, and delivery health. */
export interface NotificationDiscordPreferences {
linked: boolean;
discord_username?: string;
mode: NotificationDiscordMode;
/** Last DM delivery failure, surfaced as link health. Empty when healthy. */
link_failure?: string;
}
export interface NotificationDiscordLinkInit {
url: string;
}
export interface WebPushSubscriptionView {
id: string;
endpoint: string;
+1
View File
@@ -227,6 +227,7 @@ export const notificationKeys = {
unreadCount: () => ["notifications", "unread-count"] as const,
preferences: () => ["notifications", "preferences"] as const,
emailPreferences: () => ["notifications", "email-preferences"] as const,
discordPreferences: () => ["notifications", "discord-preferences"] as const,
capability: () => ["notifications", "capability"] as const,
webhooks: () => ["notifications", "webhooks"] as const,
webPushSubscriptions: () => ["notifications", "web-push-subscriptions"] as const,
+53
View File
@@ -3,6 +3,9 @@ import { useInfiniteQuery, useMutation, useQuery, useQueryClient } from "@tansta
import { api } from "@/api/client";
import type {
AppNotification,
NotificationDiscordLinkInit,
NotificationDiscordMode,
NotificationDiscordPreferences,
NotificationEmailPreferences,
NotificationListResponse,
NotificationPreferences,
@@ -119,6 +122,56 @@ export function useUpdateEmailNotificationPreferences() {
});
}
export function useDiscordNotificationPreferences(enabled = true) {
return useQuery({
queryKey: notificationKeys.discordPreferences(),
queryFn: () => api<NotificationDiscordPreferences>("/notifications/discord-preferences"),
enabled,
});
}
export function useUpdateDiscordNotificationPreferences() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: (update: { mode: NotificationDiscordMode }) =>
api<NotificationDiscordPreferences>("/notifications/discord-preferences", {
method: "PUT",
body: JSON.stringify(update),
}),
onSuccess: (prefs) => {
queryClient.setQueryData(notificationKeys.discordPreferences(), prefs);
},
onError: (error) => {
toast.error(error instanceof Error ? error.message : "Failed to save Discord preferences");
},
});
}
/** Starts the Discord account-link OAuth flow; navigate to the returned URL. */
export function useDiscordLinkInit() {
return useMutation({
mutationFn: () =>
api<NotificationDiscordLinkInit>("/notifications/discord/link/init", { method: "POST" }),
onError: (error) => {
toast.error(error instanceof Error ? error.message : "Failed to start Discord link");
},
});
}
export function useUnlinkDiscord() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: () => api("/notifications/discord-link", { method: "DELETE" }),
onSuccess: () => {
void queryClient.invalidateQueries({ queryKey: notificationKeys.discordPreferences() });
toast.success("Discord account unlinked");
},
onError: () => {
toast.error("Failed to unlink Discord account");
},
});
}
// --- Realtime cache reducers (used by RealtimeEventsProvider) ---
type NotificationsInfiniteData = {
+3
View File
@@ -47,6 +47,8 @@ export function useSettingsForm({ keys }: UseSettingsFormOptions) {
const dirtyCount = dirty.size;
const isDirty = useCallback((key: string) => dirty.has(key), [dirty]);
const buildConnectionCheckRequest = useCallback(
(selectedKeys: string[] = keys): AdminSettingsConnectionCheckRequest => ({
values: Object.fromEntries(
@@ -89,6 +91,7 @@ export function useSettingsForm({ keys }: UseSettingsFormOptions) {
getValue,
setValue,
dirtyCount,
isDirty,
save,
discard,
isSaving: updateSetting.isPending,
@@ -1,5 +1,8 @@
import { useMemo } from "react";
import { TriangleAlert } from "lucide-react";
import { useMemo, useState } from "react";
import { Check, Copy, ExternalLink, Loader2, Send, TriangleAlert } from "lucide-react";
import { toast } from "sonner";
import { api } from "@/api/client";
import { Button } from "@/components/ui/button";
import { useSettingsForm } from "@/hooks/useSettingsForm";
import { FieldGroup } from "./FieldGroup";
import { SaveBar } from "./SaveBar";
@@ -21,11 +24,127 @@ const KEYS = [
"notifications.email.allow_per_episode",
"notifications.email.digest_hour",
"notifications.email.external_url",
"notifications.discord_enabled",
"notifications.discord.allow_per_episode",
"notifications.discord.digest_hour",
"discord.client_id",
"discord.client_secret",
"discord.bot_token",
"notifications.retention.read_days",
"notifications.retention.unread_days",
"notifications.retention.event_days",
];
interface DiscordTestResult {
ok: boolean;
duration_ms: number;
message?: string;
}
/**
* Invite link for adding the bot to a Discord server. Membership alone is
* enough to DM, so no permissions are requested.
*/
function discordInviteUrl(clientId: string): string {
return `https://discord.com/oauth2/authorize?client_id=${encodeURIComponent(clientId)}&scope=bot&permissions=0`;
}
function InviteBotRow({ clientId }: { clientId: string }) {
const [copied, setCopied] = useState(false);
const trimmed = clientId.trim();
return (
<div className="space-y-2 py-2">
<div className="flex flex-wrap gap-1.5">
<Button
variant="outline"
size="sm"
disabled={!trimmed}
onClick={() => window.open(discordInviteUrl(trimmed), "_blank", "noopener,noreferrer")}
>
<ExternalLink className="mr-1.5 h-3.5 w-3.5" />
Invite bot to server
</Button>
<Button
variant="outline"
size="sm"
disabled={!trimmed}
onClick={() => {
void navigator.clipboard.writeText(discordInviteUrl(trimmed));
setCopied(true);
toast.success("Invite link copied");
window.setTimeout(() => setCopied(false), 2000);
}}
>
{copied ? (
<Check className="mr-1.5 h-3.5 w-3.5" />
) : (
<Copy className="mr-1.5 h-3.5 w-3.5" />
)}
Copy link
</Button>
</div>
<div className="text-muted-foreground text-xs">
{trimmed
? "Anyone with Manage Server permission on your Discord server can open this link to add the bot. Users must be members of that server to receive DMs."
: "Enter the Client ID above to generate the invite link."}
</div>
</div>
);
}
function TestDiscordRow({ unsaved }: { unsaved: boolean }) {
const [pending, setPending] = useState(false);
const [result, setResult] = useState<DiscordTestResult | null>(null);
const runTest = async () => {
setPending(true);
setResult(null);
try {
const response = await api<DiscordTestResult>("/admin/notifications/discord/test", {
method: "POST",
});
setResult(response);
if (response.ok) {
toast.success("Discord bot connected");
}
} catch (error) {
toast.error(error instanceof Error ? error.message : "Test request failed");
} finally {
setPending(false);
}
};
return (
<div className="space-y-2 py-2">
<Button
variant="outline"
size="sm"
disabled={pending || unsaved}
onClick={() => void runTest()}
>
{pending ? (
<Loader2 className="mr-1.5 h-3.5 w-3.5 animate-spin" />
) : (
<Send className="mr-1.5 h-3.5 w-3.5" />
)}
Test bot token
</Button>
{unsaved && (
<div className="text-muted-foreground text-xs">
Save your changes first — the test uses the saved credentials.
</div>
)}
{result && (
<div className={`text-xs ${result.ok ? "text-emerald-500" : "text-amber-500"}`}>
{result.ok ? "Success" : "Failed"} ({result.duration_ms}ms)
{result.message ? ` — ${result.message}` : ""}
</div>
)}
</div>
);
}
export default function NotificationsAdminSettings() {
const form = useSettingsForm({ keys: useMemo(() => KEYS, []) });
@@ -40,6 +159,16 @@ export default function NotificationsAdminSettings() {
const allowPrivate =
form.getValue("notifications.webhooks.allow_private_destinations") === "true";
// Discord is opt-in (default off); the rest of its settings only appear
// once the bot is enabled.
const discordEnabled = form.getValue("notifications.discord_enabled") === "true";
// The test endpoint reads the SAVED credentials; testing with unsaved
// edits in the form would silently test the old values.
const discordCredentialsDirty = [
"discord.client_id",
"discord.client_secret",
"discord.bot_token",
].some(form.isDirty);
return (
<div className="flex h-full flex-col">
@@ -180,6 +309,89 @@ export default function NotificationsAdminSettings() {
/>
</FieldGroup>
<FieldGroup label="Discord">
<SettingField
label="Discord Bot"
hint="Let users link their Discord account and receive notifications as bot DMs. Off by default; while off, Discord notifications are hidden everywhere."
type="toggle"
value={form.getValue("notifications.discord_enabled") || "false"}
onChange={(v) => form.setValue("notifications.discord_enabled", v)}
/>
{discordEnabled && (
<>
<div className="text-muted-foreground space-y-1.5 py-2 text-xs leading-relaxed">
<p>Set up at discord.com/developers/applications:</p>
<ol className="list-decimal space-y-1.5 pl-4">
<li>Create an application (or open an existing one).</li>
<li>
OAuth2 page: copy the <strong>Client ID</strong>, reset and copy the{" "}
<strong>Client Secret</strong>, and under Redirects add
<code className="bg-muted mx-1 rounded px-1">
{"<public URL>"}/api/v1/notifications/discord/link/callback
</code>
using this server&apos;s public URL (SILO_PUBLIC_URL) — it must match exactly.
</li>
<li>
Bot page: reset and copy the <strong>Token</strong>. Leave all Privileged
Gateway Intents (Presence, Server Members, Message Content) <strong>off</strong>{" "}
— Silo never connects to the gateway; it only sends DMs.
</li>
<li>
Keep <strong>Requires OAuth2 Code Grant</strong> off, or the invite link below
won&apos;t work. Enable <strong>Public Bot</strong> only if someone other than
the application owner will be inviting it.
</li>
<li>
Paste the three credentials below, save, then use the invite button to add the
bot to your Discord server. It needs <strong>no role permissions</strong> —
membership alone lets it DM members, and users must share that server with the
bot to receive DMs.
</li>
</ol>
</div>
<SettingField
label="Allow Per-Episode DMs"
hint="Let users choose a DM per episode instead of the daily digest. Off coerces those accounts to the digest."
type="toggle"
value={toggleValue("notifications.discord.allow_per_episode")}
onChange={(v) => form.setValue("notifications.discord.allow_per_episode", v)}
/>
<SettingField
label="Digest Hour"
hint="Hour of day (0-23, server time) when daily digest DMs go out (default 8)"
type="number"
value={numberValue("notifications.discord.digest_hour", "8")}
onChange={(v) => form.setValue("notifications.discord.digest_hour", v)}
/>
<SettingField
label="Client ID"
hint="The Discord application's OAuth2 client ID (used for account linking)"
type="text"
value={form.getValue("discord.client_id")}
onChange={(v) => form.setValue("discord.client_id", v)}
/>
<InviteBotRow clientId={form.getValue("discord.client_id")} />
<SettingField
label="Client Secret"
hint="The Discord application's OAuth2 client secret"
type="password"
value={form.getValue("discord.client_secret")}
sensitiveConfigured={form.sensitiveConfigured.includes("discord.client_secret")}
onChange={(v) => form.setValue("discord.client_secret", v)}
/>
<SettingField
label="Bot Token"
hint="The bot user's token (used to send DMs)"
type="password"
value={form.getValue("discord.bot_token")}
sensitiveConfigured={form.sensitiveConfigured.includes("discord.bot_token")}
onChange={(v) => form.setValue("discord.bot_token", v)}
/>
<TestDiscordRow unsaved={discordCredentialsDirty} />
</>
)}
</FieldGroup>
<FieldGroup label="Retention">
<SettingField
label="Read Notifications (days)"
+217 -29
View File
@@ -1,4 +1,5 @@
import { useEffect, useState } from "react";
import { useSearchParams } from "react-router";
import { useQueryClient } from "@tanstack/react-query";
import {
AlertTriangle,
@@ -16,7 +17,7 @@ import {
} from "lucide-react";
import { toast } from "sonner";
import type {
NotificationEmailMode,
NotificationChannelMode,
NotificationPreferences,
NotificationWebhook,
NotificationWebhookInput,
@@ -48,8 +49,12 @@ import { Skeleton } from "@/components/ui/skeleton";
import { Switch } from "@/components/ui/switch";
import { useAuth } from "@/hooks/useAuth";
import {
useDiscordLinkInit,
useDiscordNotificationPreferences,
useEmailNotificationPreferences,
useNotificationPreferences,
useUnlinkDiscord,
useUpdateDiscordNotificationPreferences,
useUpdateEmailNotificationPreferences,
useUpdateNotificationPreferences,
} from "@/hooks/queries/notifications";
@@ -152,6 +157,65 @@ function PreferencesSection() {
);
}
/**
* Frequency picker shared by the account-level digest channels (email,
* Discord). Per-episode stays listed-but-disabled when the admin allowance is
* off and the account is still set to it, so the coercion is visible.
*/
function ChannelFrequencyRow({
mode,
allowPerEpisode,
digestHour,
isPending,
perEpisodeHint,
onChange,
}: {
mode: NotificationChannelMode;
allowPerEpisode: boolean;
digestHour: string;
isPending: boolean;
perEpisodeHint: string;
onChange: (mode: NotificationChannelMode) => void;
}) {
const digestText = `One summary per day, around ${digestHour}:00 server time`;
return (
<div className="flex items-center justify-between gap-3">
<div>
<div className="text-sm">Frequency</div>
<div className="text-muted-foreground text-xs">
{mode === "per_episode"
? perEpisodeHint
: mode === "per_episode_and_digest"
? `${perEpisodeHint}, plus a daily summary around ${digestHour}:00 server time`
: digestText}
</div>
</div>
<Select
value={mode}
disabled={isPending}
onValueChange={(value) => onChange(value as NotificationChannelMode)}
>
<SelectTrigger className="w-[220px]">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="daily_digest">Daily digest</SelectItem>
{(allowPerEpisode || mode === "per_episode") && (
<SelectItem value="per_episode" disabled={!allowPerEpisode}>
Every episode
</SelectItem>
)}
{(allowPerEpisode || mode === "per_episode_and_digest") && (
<SelectItem value="per_episode_and_digest" disabled={!allowPerEpisode}>
Every episode + daily digest
</SelectItem>
)}
</SelectContent>
</Select>
</div>
);
}
function EmailSection() {
const { user } = useAuth();
const capability = useNotificationCapability();
@@ -198,35 +262,16 @@ function EmailSection() {
/>
</div>
{enabled && (
<div className="flex items-center justify-between gap-3">
<div>
<div className="text-sm">Frequency</div>
<div className="text-muted-foreground text-xs">
{mode === "per_episode"
? "An email as soon as each notification arrives"
: `One summary per day, around ${digestHour}:00 server time`}
</div>
</div>
<Select
value={mode}
disabled={updatePrefs.isPending}
onValueChange={(value) => updatePrefs.mutate({ mode: value as NotificationEmailMode })}
>
<SelectTrigger className="w-[180px]">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="daily_digest">Daily digest</SelectItem>
{(allowPerEpisode || mode === "per_episode") && (
<SelectItem value="per_episode" disabled={!allowPerEpisode}>
Every episode
</SelectItem>
)}
</SelectContent>
</Select>
</div>
<ChannelFrequencyRow
mode={mode}
allowPerEpisode={allowPerEpisode}
digestHour={digestHour}
isPending={updatePrefs.isPending}
perEpisodeHint="An email as soon as each notification arrives"
onChange={(value) => updatePrefs.mutate({ mode: value })}
/>
)}
{enabled && mode === "per_episode" && !allowPerEpisode && (
{enabled && mode !== "daily_digest" && !allowPerEpisode && (
<div className="text-xs text-amber-500">
Per-episode email is disabled by the administrator; you'll receive the daily digest
instead.
@@ -236,6 +281,147 @@ function EmailSection() {
);
}
const DISCORD_LINK_ERRORS: Record<string, string> = {
denied: "Discord linking was cancelled",
disabled: "Discord notifications were disabled by the administrator",
invalid_callback: "Discord linking failed: invalid callback",
state_invalid: "Discord linking expired — try again",
exchange_failed: "Discord linking failed — try again",
};
function DiscordSection() {
const queryClient = useQueryClient();
const [searchParams, setSearchParams] = useSearchParams();
const capability = useNotificationCapability();
const discordCap = capability.data?.discord;
const available = discordCap?.available ?? false;
const { data: prefs, isLoading } = useDiscordNotificationPreferences(available);
const updatePrefs = useUpdateDiscordNotificationPreferences();
const linkInit = useDiscordLinkInit();
const unlink = useUnlinkDiscord();
// Surface the OAuth callback result (the link flow round-trips through
// Discord and lands back here with a query param), then clean the URL.
useEffect(() => {
const linked = searchParams.get("discord_linked");
const error = searchParams.get("discord_error");
if (!linked && !error) {
return;
}
if (linked) {
toast.success("Discord account linked");
void queryClient.invalidateQueries({ queryKey: notificationKeys.discordPreferences() });
} else if (error) {
toast.error(DISCORD_LINK_ERRORS[error] ?? "Discord linking failed");
}
const next = new URLSearchParams(searchParams);
next.delete("discord_linked");
next.delete("discord_error");
setSearchParams(next, { replace: true });
}, [searchParams, setSearchParams, queryClient]);
if (!available) {
return null;
}
if (isLoading) {
return (
<SettingsGroup title="Discord Notifications">
<Skeleton className="h-16 w-full" />
</SettingsGroup>
);
}
const linked = prefs?.linked ?? false;
const mode = prefs?.mode ?? "off";
const enabled = mode !== "off";
const allowPerEpisode = discordCap?.modes.includes("per_episode") ?? false;
const digestHour = String(discordCap?.digest_hour ?? 8).padStart(2, "0");
const startLink = () => {
linkInit.mutate(undefined, {
onSuccess: (init) => {
window.location.assign(init.url);
},
});
};
return (
<SettingsGroup
title="Discord Notifications"
description="Account-wide: the Silo bot sends direct messages covering every profile on this account. You must share a Discord server with the bot."
>
{!linked ? (
<div className="flex items-center justify-between gap-3">
<div>
<div className="text-sm font-medium">Link your Discord account</div>
<div className="text-muted-foreground text-xs">
Authorize on Discord so the bot knows who to message
</div>
</div>
<Button size="sm" disabled={linkInit.isPending} onClick={startLink}>
{linkInit.isPending && <Loader2 className="mr-1.5 h-3.5 w-3.5 animate-spin" />}
Link Discord
</Button>
</div>
) : (
<>
<div className="flex items-center justify-between gap-3">
<div>
<div className="text-sm font-medium">
Linked as {prefs?.discord_username || "Discord user"}
</div>
<div className="text-muted-foreground text-xs">
Notifications you'd see in the inbox, delivered as a DM
</div>
</div>
<div className="flex items-center gap-2">
<Switch
checked={enabled}
disabled={updatePrefs.isPending}
onCheckedChange={(checked) =>
updatePrefs.mutate({ mode: checked ? "daily_digest" : "off" })
}
/>
<Button
variant="ghost"
size="sm"
className="text-destructive"
disabled={unlink.isPending}
onClick={() => unlink.mutate()}
>
Unlink
</Button>
</div>
</div>
{enabled && (
<ChannelFrequencyRow
mode={mode}
allowPerEpisode={allowPerEpisode}
digestHour={digestHour}
isPending={updatePrefs.isPending}
perEpisodeHint="A DM as soon as each notification arrives"
onChange={(value) => updatePrefs.mutate({ mode: value })}
/>
)}
{enabled && mode !== "daily_digest" && !allowPerEpisode && (
<div className="text-xs text-amber-500">
Per-episode DMs are disabled by the administrator; you'll receive the daily digest
instead.
</div>
)}
{enabled && prefs?.link_failure && (
<div className="flex items-start gap-1.5 text-xs text-amber-500">
<AlertTriangle className="mt-0.5 h-3.5 w-3.5 shrink-0" />
<span>{prefs.link_failure}</span>
</div>
)}
</>
)}
</SettingsGroup>
);
}
/**
* Delivery health for one push subscription, derived the same way as webhook
* health: a failure newer than the last success means the device is failing.
@@ -829,6 +1015,8 @@ export default function NotificationsSettings() {
<EmailSection />
<DiscordSection />
<WebhooksSection />
</div>
);