diff --git a/.env.example b/.env.example index b8908d72..e699a644 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/internal/api/handlers/notifications.go b/internal/api/handlers/notifications.go index 6c9279e3..7a4d351c 100644 --- a/internal/api/handlers/notifications.go +++ b/internal/api/handlers/notifications.go @@ -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, }) } diff --git a/internal/api/handlers/notifications_discord.go b/internal/api/handlers/notifications_discord.go new file mode 100644 index 00000000..c9a12333 --- /dev/null +++ b/internal/api/handlers/notifications_discord.go @@ -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= 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, + }) +} diff --git a/internal/api/router.go b/internal/api/router.go index d263213c..b7e4c33c 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -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) diff --git a/internal/catalog/encrypted_settings_repo.go b/internal/catalog/encrypted_settings_repo.go index 610a1662..0c9e67a2 100644 --- a/internal/catalog/encrypted_settings_repo.go +++ b/internal/catalog/encrypted_settings_repo.go @@ -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). diff --git a/internal/discord/client.go b/internal/discord/client.go new file mode 100644 index 00000000..a0850ae1 --- /dev/null +++ b/internal/discord/client.go @@ -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) +} diff --git a/internal/discord/client_test.go b/internal/discord/client_test.go new file mode 100644 index 00000000..0d4c3c61 --- /dev/null +++ b/internal/discord/client_test.go @@ -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") + } +} diff --git a/internal/notifications/account_channel_engine.go b/internal/notifications/account_channel_engine.go new file mode 100644 index 00000000..076f20ff --- /dev/null +++ b/internal/notifications/account_channel_engine.go @@ -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 +} diff --git a/internal/notifications/delivery_repo.go b/internal/notifications/delivery_repo.go index cc545d9e..c3cf6c58 100644 --- a/internal/notifications/delivery_repo.go +++ b/internal/notifications/delivery_repo.go @@ -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, diff --git a/internal/notifications/discord_dm.go b/internal/notifications/discord_dm.go new file mode 100644 index 00000000..eb14b24d --- /dev/null +++ b/internal/notifications/discord_dm.go @@ -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) +} diff --git a/internal/notifications/discord_prefs_repo.go b/internal/notifications/discord_prefs_repo.go new file mode 100644 index 00000000..1ff6e45d --- /dev/null +++ b/internal/notifications/discord_prefs_repo.go @@ -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 +} diff --git a/internal/notifications/email_digest.go b/internal/notifications/email_digest.go index 110cd817..d99dde64 100644 --- a/internal/notifications/email_digest.go +++ b/internal/notifications/email_digest.go @@ -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 -} diff --git a/internal/notifications/email_logic_test.go b/internal/notifications/email_logic_test.go index 064d693e..d56a8f19 100644 --- a/internal/notifications/email_logic_test.go +++ b/internal/notifications/email_logic_test.go @@ -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") } } diff --git a/internal/notifications/email_prefs_repo.go b/internal/notifications/email_prefs_repo.go index 5f1a7378..71046486 100644 --- a/internal/notifications/email_prefs_repo.go +++ b/internal/notifications/email_prefs_repo.go @@ -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 -} diff --git a/internal/notifications/settings.go b/internal/notifications/settings.go index bbcd345b..ca6a23f9 100644 --- a/internal/notifications/settings.go +++ b/internal/notifications/settings.go @@ -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)) +} diff --git a/internal/notifications/system.go b/internal/notifications/system.go index f8897ba7..2c441b80 100644 --- a/internal/notifications/system.go +++ b/internal/notifications/system.go @@ -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 } diff --git a/internal/notifications/webhook_payload_discord.go b/internal/notifications/webhook_payload_discord.go index 86090d0b..7edf4ab6 100644 --- a/internal/notifications/webhook_payload_discord.go +++ b/internal/notifications/webhook_payload_discord.go @@ -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 { diff --git a/migrations/sql/20260611223317_discord_notification_channel.sql b/migrations/sql/20260611223317_discord_notification_channel.sql new file mode 100644 index 00000000..eefb54fe --- /dev/null +++ b/migrations/sql/20260611223317_discord_notification_channel.sql @@ -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 diff --git a/migrations/sql/20260611234431_account_channel_combined_mode.sql b/migrations/sql/20260611234431_account_channel_combined_mode.sql new file mode 100644 index 00000000..60c0247e --- /dev/null +++ b/migrations/sql/20260611234431_account_channel_combined_mode.sql @@ -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 diff --git a/web/src/api/types.ts b/web/src/api/types.ts index ae227446..174400b5 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -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; diff --git a/web/src/hooks/queries/keys.ts b/web/src/hooks/queries/keys.ts index 12382ccc..b074d478 100644 --- a/web/src/hooks/queries/keys.ts +++ b/web/src/hooks/queries/keys.ts @@ -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, diff --git a/web/src/hooks/queries/notifications.ts b/web/src/hooks/queries/notifications.ts index 2c4e48c4..1f6b0cc9 100644 --- a/web/src/hooks/queries/notifications.ts +++ b/web/src/hooks/queries/notifications.ts @@ -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("/notifications/discord-preferences"), + enabled, + }); +} + +export function useUpdateDiscordNotificationPreferences() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (update: { mode: NotificationDiscordMode }) => + api("/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("/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 = { diff --git a/web/src/hooks/useSettingsForm.ts b/web/src/hooks/useSettingsForm.ts index beae94c8..120a3dfa 100644 --- a/web/src/hooks/useSettingsForm.ts +++ b/web/src/hooks/useSettingsForm.ts @@ -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, diff --git a/web/src/pages/admin-settings/NotificationsAdminSettings.tsx b/web/src/pages/admin-settings/NotificationsAdminSettings.tsx index 380f5820..7a94956e 100644 --- a/web/src/pages/admin-settings/NotificationsAdminSettings.tsx +++ b/web/src/pages/admin-settings/NotificationsAdminSettings.tsx @@ -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 ( +
+
+ + +
+
+ {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."} +
+
+ ); +} + +function TestDiscordRow({ unsaved }: { unsaved: boolean }) { + const [pending, setPending] = useState(false); + const [result, setResult] = useState(null); + + const runTest = async () => { + setPending(true); + setResult(null); + try { + const response = await api("/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 ( +
+ + {unsaved && ( +
+ Save your changes first — the test uses the saved credentials. +
+ )} + {result && ( +
+ {result.ok ? "Success" : "Failed"} ({result.duration_ms}ms) + {result.message ? ` — ${result.message}` : ""} +
+ )} +
+ ); +} + 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 (
@@ -180,6 +309,89 @@ export default function NotificationsAdminSettings() { /> + + form.setValue("notifications.discord_enabled", v)} + /> + {discordEnabled && ( + <> +
+

Set up at discord.com/developers/applications:

+
    +
  1. Create an application (or open an existing one).
    2. +
  2. + OAuth2 page: copy the Client ID, reset and copy the{" "} + Client Secret, and under Redirects add + + {""}/api/v1/notifications/discord/link/callback + + using this server's public URL (SILO_PUBLIC_URL) — it must match exactly. +
    3. +
  3. + Bot page: reset and copy the Token. Leave all Privileged + Gateway Intents (Presence, Server Members, Message Content) off{" "} + — Silo never connects to the gateway; it only sends DMs. +
    4. +
  4. + Keep Requires OAuth2 Code Grant off, or the invite link below + won't work. Enable Public Bot only if someone other than + the application owner will be inviting it. +
    5. +
  5. + Paste the three credentials below, save, then use the invite button to add the + bot to your Discord server. It needs no role permissions — + membership alone lets it DM members, and users must share that server with the + bot to receive DMs. +
    6. +
+
+ form.setValue("notifications.discord.allow_per_episode", v)} + /> + form.setValue("notifications.discord.digest_hour", v)} + /> + form.setValue("discord.client_id", v)} + /> + + form.setValue("discord.client_secret", v)} + /> + form.setValue("discord.bot_token", v)} + /> + + + )} + + void; +}) { + const digestText = `One summary per day, around ${digestHour}:00 server time`; + return ( +
+
+
Frequency
+
+ {mode === "per_episode" + ? perEpisodeHint + : mode === "per_episode_and_digest" + ? `${perEpisodeHint}, plus a daily summary around ${digestHour}:00 server time` + : digestText} +
+
+ +
+ ); +} + function EmailSection() { const { user } = useAuth(); const capability = useNotificationCapability(); @@ -198,35 +262,16 @@ function EmailSection() { />
{enabled && ( -
-
-
Frequency
-
- {mode === "per_episode" - ? "An email as soon as each notification arrives" - : `One summary per day, around ${digestHour}:00 server time`} -
-
- -
+ updatePrefs.mutate({ mode: value })} + /> )} - {enabled && mode === "per_episode" && !allowPerEpisode && ( + {enabled && mode !== "daily_digest" && !allowPerEpisode && (
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 = { + 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 ( + + + + ); + } + + 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 ( + + {!linked ? ( +
+
+
Link your Discord account
+
+ Authorize on Discord so the bot knows who to message +
+
+ +
+ ) : ( + <> +
+
+
+ Linked as {prefs?.discord_username || "Discord user"} +
+
+ Notifications you'd see in the inbox, delivered as a DM +
+
+
+ + updatePrefs.mutate({ mode: checked ? "daily_digest" : "off" }) + } + /> + +
+
+ {enabled && ( + updatePrefs.mutate({ mode: value })} + /> + )} + {enabled && mode !== "daily_digest" && !allowPerEpisode && ( +
+ Per-episode DMs are disabled by the administrator; you'll receive the daily digest + instead. +
+ )} + {enabled && prefs?.link_failure && ( +
+ + {prefs.link_failure} +
+ )} + + )} + + ); +} + /** * 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() { + +
);