From e5b210589d76faa54ec6811433f92cdf4b6fc0ed Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Thu, 11 Jun 2026 17:16:27 -0400 Subject: [PATCH] fix(notifications): address PR #136 review findings Codex + CodeRabbit review fixes, all verified against current behavior: - Web Push: single-writer VAPID provisioning via a new conditional SetIfAbsent settings write (no split-brain identity across nodes), and read/decode failures now surface instead of silently rotating the keypair; the eager-provisioning goroutine joins the shutdown WaitGroup - Web Push: endpoint reassignment purges the previous owner's pending attempts inside the upsert transaction, with an ownership re-check at send time - Webhooks: per-profile cap enforced atomically (advisory-locked count+insert), typed pgconn unique-violation mapping, create-time type/URL mismatch rejection, send-time HTTPS re-check, and Retry-After HTTP-date support (shared, clamped parser also used by web push) - Delivery workers: transient delivery-row lookup errors leave the claim to lease expiry instead of permanently failing the attempt - Interest: history-only imports now feed the index (userstore history hooks + completed-history folding in recompute/rebuild), rebuild also recomputes existing interest rows so removed sources get cleaned up, and failed flush mutations requeue (bounded) instead of dropping - Retention: read notifications age from read_at, not created_at - Startup: scan queue workers start only after the availability detector is wired, so resumed scans cannot skip availability recording - mail: settings-store read failures propagate instead of reading as "not configured" - DB: new migration adds episode ordinal/key CHECK constraints - Web: service worker restricts notification clicks to same-origin URLs, preferences popover gets an error+retry state, and the realtime profile-rebind backoff grows to 5 minutes to keep shared channels stable through notifications-only outages Co-Authored-By: Claude Fable 5 --- cmd/silo/main.go | 14 ++- .../api/handlers/notifications_webhooks.go | 2 +- internal/catalog/encrypted_settings_repo.go | 24 ++++++ internal/catalog/server_settings_repo.go | 17 ++++ internal/mail/mail.go | 21 ++++- internal/notifications/delivery_repo.go | 7 +- internal/notifications/interest_hooks.go | 43 +++++++++- internal/notifications/interest_repo.go | 23 +++++ internal/notifications/interest_updater.go | 86 ++++++++++++++++--- internal/notifications/system.go | 34 +++++++- internal/notifications/webhook_guard.go | 2 +- internal/notifications/webhook_http.go | 36 +++++++- .../notifications/webhook_payload_discord.go | 2 +- internal/notifications/webhook_repo.go | 60 +++++++++---- internal/notifications/webhook_sender.go | 14 ++- internal/notifications/webhook_service.go | 23 ++--- internal/notifications/webhook_types.go | 7 ++ internal/notifications/webpush_repo.go | 25 +++++- internal/notifications/webpush_sender.go | 25 ++++-- internal/notifications/webpush_service.go | 69 ++++++++++----- .../20260611202951_episode_ordinal_checks.sql | 33 +++++++ web/public/sw.js | 13 ++- web/src/components/RealtimeEventsProvider.tsx | 9 +- web/src/pages/Notifications.tsx | 11 ++- 24 files changed, 507 insertions(+), 93 deletions(-) create mode 100644 migrations/sql/20260611202951_episode_ordinal_checks.sql diff --git a/cmd/silo/main.go b/cmd/silo/main.go index c0a3fbaf..3169f7b7 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1219,8 +1219,10 @@ func main() { cfg.Scanner.MaxConcurrentLibraries, cfg.Scanner.MaxConcurrentScoped, ) - libraryScanQueue.Start() - defer libraryScanQueue.Stop() + // Started below, after the notification system has attached its + // availability detector to the executor: a scan resumed by the + // workers before that wiring would complete without recording + // episode availability, silently losing release notifications. deps.LibraryScanQueue = libraryScanQueue } if deps.DB != nil && deps.FileRepo != nil && metadataService != nil { @@ -1315,6 +1317,14 @@ func main() { } } + // Start the scan queue only now that the availability detector (when + // notifications are enabled) is attached to the ingest executor, so scans + // resumed at startup cannot complete before the detector exists. + if libraryScanQueue != nil { + libraryScanQueue.Start() + defer libraryScanQueue.Stop() + } + if userStoreProvider != nil && pluginService != nil { deps.PluginUserConfig = plugins.NewUserConfigStore(userStoreProvider, pluginService) } diff --git a/internal/api/handlers/notifications_webhooks.go b/internal/api/handlers/notifications_webhooks.go index c052d7a9..e009b114 100644 --- a/internal/api/handlers/notifications_webhooks.go +++ b/internal/api/handlers/notifications_webhooks.go @@ -179,7 +179,7 @@ func (h *NotificationsHandler) HandleDeleteWebhook(w http.ResponseWriter, r *htt } profileID := apimw.GetProfileID(r.Context()) if err := service.Delete(r.Context(), profileID, chi.URLParam(r, "id")); err != nil { - writeError(w, http.StatusInternalServerError, "internal_error", "Failed to delete webhook") + writeWebhookError(w, err) return } w.WriteHeader(http.StatusNoContent) diff --git a/internal/catalog/encrypted_settings_repo.go b/internal/catalog/encrypted_settings_repo.go index 5185644f..610a1662 100644 --- a/internal/catalog/encrypted_settings_repo.go +++ b/internal/catalog/encrypted_settings_repo.go @@ -135,6 +135,30 @@ func (r *EncryptedSettingsRepo) Set(ctx context.Context, key, value string) erro return r.inner.Set(ctx, key, value) } +// settingsConditionalWriter is the optional conditional-write capability of a +// raw settings store (satisfied by *ServerSettingsRepo). +type settingsConditionalWriter interface { + SetIfAbsent(ctx context.Context, key, value string) (bool, error) +} + +// SetIfAbsent applies Set's encryption contract to a conditional write: the +// value lands only when the key currently has no value, so concurrent +// provisioners of generated secrets cannot overwrite each other. +func (r *EncryptedSettingsRepo) SetIfAbsent(ctx context.Context, key, value string) (bool, error) { + inner, ok := r.inner.(settingsConditionalWriter) + if !ok { + return false, fmt.Errorf("settings store does not support conditional writes") + } + if SensitiveSettingKeys[key] && value != "" { + ct, err := r.cipher.Encrypt(value, secret.SettingsAAD(key)) + if err != nil { + return false, fmt.Errorf("encrypt setting %q: %w", key, err) + } + value = ct + } + return inner.SetIfAbsent(ctx, key, value) +} + // Get reads a value and applies the read-path contract: legacy plaintext passes // through, an enc:v1: value is decrypted, and a corrupt ciphertext errors. func (r *EncryptedSettingsRepo) Get(ctx context.Context, key string) (string, error) { diff --git a/internal/catalog/server_settings_repo.go b/internal/catalog/server_settings_repo.go index 104032c7..0ad114f3 100644 --- a/internal/catalog/server_settings_repo.go +++ b/internal/catalog/server_settings_repo.go @@ -45,6 +45,23 @@ func (r *ServerSettingsRepo) Set(ctx context.Context, key, value string) error { return nil } +// SetIfAbsent inserts a setting only when the key has no value yet (absent or +// empty), reporting whether this call won the write. Generated credentials +// (e.g. the web push VAPID keypair) must be provisioned single-writer across +// concurrent nodes: exactly one generated value may ever land. +func (r *ServerSettingsRepo) SetIfAbsent(ctx context.Context, key, value string) (bool, error) { + tag, err := r.pool.Exec(ctx, + `INSERT INTO server_settings (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value + WHERE server_settings.value = ''`, + key, value, + ) + if err != nil { + return false, fmt.Errorf("server_settings set-if-absent %q: %w", key, err) + } + return tag.RowsAffected() > 0, nil +} + // GetAll retrieves all settings as a map. func (r *ServerSettingsRepo) GetAll(ctx context.Context) (map[string]string, error) { rows, err := r.pool.Query(ctx, `SELECT key, value FROM server_settings`) diff --git a/internal/mail/mail.go b/internal/mail/mail.go index 3fda72d1..5f7a9a79 100644 --- a/internal/mail/mail.go +++ b/internal/mail/mail.go @@ -106,11 +106,22 @@ func (s *SMTPSender) loadConfig(ctx context.Context) (*smtpConfig, error) { if s == nil || s.settings == nil { return nil, ErrNotConfigured } + // A settings-store failure must surface as an error, never be mistaken + // for "email is not configured" — that would silently hide real backend + // problems behind a graceful-degradation path. + var readErr error get := func(key string) string { - value, _ := s.settings.Get(ctx, key) + value, err := s.settings.Get(ctx, key) + if err != nil && readErr == nil { + readErr = fmt.Errorf("read setting %s: %w", key, err) + } return strings.TrimSpace(value) } - if !truthy(get(SettingEnabled)) { + enabled := truthy(get(SettingEnabled)) + if readErr != nil { + return nil, readErr + } + if !enabled { return nil, ErrNotConfigured } cfg := &smtpConfig{ @@ -122,10 +133,14 @@ func (s *SMTPSender) loadConfig(ctx context.Context) (*smtpConfig, error) { fromAddress: get(SettingFromAddress), fromName: get(SettingFromName), } + portRaw := get(SettingSMTPPort) + if readErr != nil { + return nil, readErr + } if cfg.host == "" || cfg.fromAddress == "" { return nil, ErrNotConfigured } - if raw := get(SettingSMTPPort); raw != "" { + if raw := portRaw; raw != "" { port, err := strconv.Atoi(raw) if err != nil || port < 1 || port > 65535 { return nil, fmt.Errorf("invalid email.smtp_port %q", raw) diff --git a/internal/notifications/delivery_repo.go b/internal/notifications/delivery_repo.go index e5aa58b7..e114a688 100644 --- a/internal/notifications/delivery_repo.go +++ b/internal/notifications/delivery_repo.go @@ -298,12 +298,13 @@ func (r *DeliveryRepository) DeleteAllForProfile(ctx context.Context, profileID return err } -// DeleteOld applies retention: read rows past readCutoff, unread rows past -// unreadCutoff. +// DeleteOld applies retention: rows read longer ago than readCutoff, unread +// rows created before unreadCutoff. Read rows age from read_at, not +// created_at — an old notification read today starts a fresh read window. func (r *DeliveryRepository) DeleteOld(ctx context.Context, readCutoff, unreadCutoff time.Time) (int64, error) { tag, err := r.pool.Exec(ctx, ` DELETE FROM notification_deliveries - WHERE (read_at IS NOT NULL AND created_at < $1) + WHERE (read_at IS NOT NULL AND read_at < $1) OR (read_at IS NULL AND created_at < $2)`, readCutoff, unreadCutoff) if err != nil { diff --git a/internal/notifications/interest_hooks.go b/internal/notifications/interest_hooks.go index 4a793c51..6f222fe1 100644 --- a/internal/notifications/interest_hooks.go +++ b/internal/notifications/interest_hooks.go @@ -9,10 +9,11 @@ import ( ) // WrapUserStoreProvider decorates the shared user-store provider so every -// favorites, watchlist, and watch-progress mutation — regardless of which -// path performed it (REST handlers, jellycompat, history imports, playback -// stop, watch sync) — queues an interest recompute. Hooking the lowest shared -// layer keeps the seven-plus mutation call sites hook-free and drift-free. +// favorites, watchlist, watch-progress, and watch-history mutation — +// regardless of which path performed it (REST handlers, jellycompat, history +// imports, playback stop, watch sync) — queues an interest recompute. Hooking +// the lowest shared layer keeps the seven-plus mutation call sites hook-free +// and drift-free. // // Progress writes queue only on state *transitions* (a row appearing, the // in-progress flag flipping, completion crossing, rows being cleared): @@ -219,6 +220,40 @@ func (s *interestTrackingStore) ClearProgressBatch(ctx context.Context, profileI return err } +// --- Watch history: history imports and watch-provider syncs may record a +// completed watch without any progress write, so the progress hooks alone +// would never see them. AddHistory (the live playback path) is deliberately +// not hooked: playback always writes progress alongside it, and those writes +// already queue on transitions. + +func (s *interestTrackingStore) AddHistoryIfMissing(ctx context.Context, entry userstore.WatchHistoryEntry) (bool, error) { + created, err := s.UserStore.AddHistoryIfMissing(ctx, entry) + if err == nil && created && entry.Completed { + s.updater.QueueItemMutation(s.userID, entry.ProfileID, entry.MediaItemID) + } + return created, err +} + +func (s *interestTrackingStore) RemoveHistoryItems(ctx context.Context, profileID string, mediaItemIDs []string, removedAt time.Time) error { + err := s.UserStore.RemoveHistoryItems(ctx, profileID, mediaItemIDs, removedAt) + if err == nil { + for _, mediaItemID := range mediaItemIDs { + s.updater.QueueItemMutation(s.userID, profileID, mediaItemID) + } + } + return err +} + +func (s *interestTrackingStore) DeleteHistoryBySource(ctx context.Context, profileID string, mediaItemIDs []string, source userstore.WatchHistorySource) error { + err := s.UserStore.DeleteHistoryBySource(ctx, profileID, mediaItemIDs, source) + if err == nil { + for _, mediaItemID := range mediaItemIDs { + s.updater.QueueItemMutation(s.userID, profileID, mediaItemID) + } + } + return err +} + // DeleteProfile purges notification state alongside the profile itself; // profiles may live outside Postgres, so no cascade covers these tables. // The purge is best-effort: a failure is logged, never surfaced as a diff --git a/internal/notifications/interest_repo.go b/internal/notifications/interest_repo.go index 256b87f2..16f17939 100644 --- a/internal/notifications/interest_repo.go +++ b/internal/notifications/interest_repo.go @@ -121,6 +121,29 @@ func (r *InterestRepository) DeleteStaleForProfileSeries(ctx context.Context, pr return nil } +// ListSeriesForProfile returns the distinct series that currently have +// interest rows for the profile. The rebuild pass recomputes these alongside +// the series resolved from live sources so rows whose sources were removed +// while the live updater was down get cleaned up instead of lingering. +func (r *InterestRepository) ListSeriesForProfile(ctx context.Context, profileID string) ([]string, error) { + rows, err := r.pool.Query(ctx, + `SELECT DISTINCT series_id FROM profile_series_interest WHERE profile_id = $1`, + profileID) + if err != nil { + return nil, fmt.Errorf("list profile interest series: %w", err) + } + defer rows.Close() + seriesIDs := make([]string, 0, 16) + for rows.Next() { + var seriesID string + if err := rows.Scan(&seriesID); err != nil { + return nil, fmt.Errorf("scan profile interest series: %w", err) + } + seriesIDs = append(seriesIDs, seriesID) + } + return seriesIDs, rows.Err() +} + // GuardedSetLastNotified raises last_notified_episode_key for the given // profiles. The < guard makes concurrent workers handling adjacent release // events safe: the higher key wins regardless of commit order. diff --git a/internal/notifications/interest_updater.go b/internal/notifications/interest_updater.go index f2c9ba76..3811e18c 100644 --- a/internal/notifications/interest_updater.go +++ b/internal/notifications/interest_updater.go @@ -25,6 +25,10 @@ const ( interestFlushInterval = 2 * time.Second interestQueryChunk = 500 interestRecomputeTime = 30 * time.Second + // interestMaxFlushAttempts bounds requeues of a failing mutation so a + // poisoned item cannot retry every flush forever; the periodic rebuild + // task repairs whatever gets dropped. + interestMaxFlushAttempts = 5 ) type interestMutation struct { @@ -48,8 +52,10 @@ type InterestUpdater struct { scopes ScopeResolver logger *slog.Logger - mu sync.Mutex - pending map[interestMutation]struct{} + mu sync.Mutex + // pending maps each queued mutation to how many flushes have already + // failed it (transient failures requeue instead of dropping). + pending map[interestMutation]int } // NewInterestUpdater creates an InterestUpdater. @@ -65,7 +71,7 @@ func NewInterestUpdater( stores: stores, scopes: scopes, logger: slog.Default().With("component", "notifications.interest"), - pending: make(map[interestMutation]struct{}), + pending: make(map[interestMutation]int), } } @@ -77,8 +83,11 @@ func (u *InterestUpdater) QueueItemMutation(userID int, profileID, itemID string if u == nil || userID <= 0 || profileID == "" || itemID == "" { return } + mutation := interestMutation{userID: userID, profileID: profileID, itemID: itemID} u.mu.Lock() - u.pending[interestMutation{userID: userID, profileID: profileID, itemID: itemID}] = struct{}{} + if _, queued := u.pending[mutation]; !queued { + u.pending[mutation] = 0 + } u.mu.Unlock() } @@ -103,9 +112,14 @@ func (u *InterestUpdater) flush(ctx context.Context) { return } batch := u.pending - u.pending = make(map[interestMutation]struct{}) + u.pending = make(map[interestMutation]int) u.mu.Unlock() + // Transient failures requeue for the next flush instead of dropping the + // mutation, which would leave profile_series_interest stale until some + // later mutation or the rebuild task happened to touch the series. + requeue := make(map[interestMutation]int) + // Resolve items to series and dedupe to one recompute per // (user, profile, series). type recomputeKey struct { @@ -114,14 +128,15 @@ func (u *InterestUpdater) flush(ctx context.Context) { seriesID string } seen := make(map[recomputeKey]struct{}, len(batch)) - for mutation := range batch { + for mutation, failures := range batch { if ctx.Err() != nil { - return + return // shutting down; the rebuild task repairs anything dropped } seriesID, ok, err := u.resolveSeriesID(ctx, mutation.itemID) if err != nil { u.logger.Warn("interest series resolution failed", "item_id", mutation.itemID, "error", err) + requeue[mutation] = failures + 1 continue } if !ok { @@ -140,8 +155,27 @@ func (u *InterestUpdater) flush(ctx context.Context) { u.logger.Warn("interest recompute failed", "user_id", mutation.userID, "profile_id", mutation.profileID, "series_id", seriesID, "error", err) + requeue[mutation] = failures + 1 } } + + if len(requeue) == 0 { + return + } + u.mu.Lock() + for mutation, failures := range requeue { + if failures >= interestMaxFlushAttempts { + u.logger.Warn("interest mutation dropped after repeated failures", + "item_id", mutation.itemID, "profile_id", mutation.profileID) + continue + } + // A fresh queue of the same mutation (failure count 0) wins; it will + // be recomputed either way. + if _, queued := u.pending[mutation]; !queued { + u.pending[mutation] = failures + } + } + u.mu.Unlock() } // resolveSeriesID maps a media item ID (series, season, or episode) to its @@ -208,6 +242,13 @@ func (u *InterestUpdater) RecomputeSeries(ctx context.Context, userID int, profi hasProgression := false lastCompletedKey := 0 hasCompleted := false + markCompleted := func(episodeID string) { + hasProgression = true + if key, ok := episodeKeys[episodeID]; ok && (!hasCompleted || key > lastCompletedKey) { + lastCompletedKey = key + hasCompleted = true + } + } for start := 0; start < len(episodeIDs); start += interestQueryChunk { end := min(start+interestQueryChunk, len(episodeIDs)) progress, err := store.ListProgressByMediaItems(ctx, profileID, episodeIDs[start:end]) @@ -220,14 +261,37 @@ func (u *InterestUpdater) RecomputeSeries(ctx context.Context, userID int, profi continueWatching = true } if entry.Completed { - if key, ok := episodeKeys[episodeID]; ok && (!hasCompleted || key > lastCompletedKey) { - lastCompletedKey = key - hasCompleted = true - } + markCompleted(episodeID) } } } + // Completed watch history counts too: history imports and watch-provider + // syncs may record a watched-at fact without ever writing a progress row, + // and those episodes must still advance the progression cursor. + for start := 0; start < len(episodeIDs); start += interestQueryChunk { + end := min(start+interestQueryChunk, len(episodeIDs)) + chunk := episodeIDs[start:end] + for offset := 0; ; { + entries, err := store.ListCompletedHistory(ctx, userstore.CompletedHistoryQuery{ + ProfileID: profileID, + MediaItemIDs: chunk, + Limit: interestQueryChunk, + Offset: offset, + }) + if err != nil { + return fmt.Errorf("load completed history: %w", err) + } + for _, entry := range entries { + markCompleted(entry.MediaItemID) + } + if len(entries) < interestQueryChunk { + break + } + offset += len(entries) + } + } + // Conservative progression cursor: next_expected = last completed key + 1. // When the profile has gaps (completed E10 with E05 unwatched), a late- // arriving E05 will not match next_up (it may still match favorite / diff --git a/internal/notifications/system.go b/internal/notifications/system.go index dfe77961..983e3789 100644 --- a/internal/notifications/system.go +++ b/internal/notifications/system.go @@ -227,7 +227,9 @@ func (s *System) Start(ctx context.Context) { // Provision the VAPID keypair eagerly so a broken settings store // surfaces at startup instead of on the first subscribe, and the // capability endpoint never pays the generation latency. + s.wg.Add(1) go func() { + defer s.wg.Done() if _, err := s.WebPush.PublicKey(ctx); err != nil && ctx.Err() == nil { s.logger.Error("web push VAPID provisioning failed", "error", err) } @@ -419,7 +421,10 @@ func (s *System) RebuildInterest(ctx context.Context, progress func(percent int, } // rebuildProfileInterest recomputes every series the profile has any -// relationship with (favorites, watchlist, progress). +// relationship with (favorites, watchlist, progress, completed watch +// history), plus every series that already has interest rows in Postgres — +// rows whose sources were all removed must be recomputed too, or the +// drift-repair pass would keep notifying about unfollowed shows forever. func (s *System) rebuildProfileInterest(ctx context.Context, store userstore.UserStore, userID int, profileID string) error { const pageSize = 500 itemIDs := make(map[string]struct{}, 64) @@ -460,14 +465,37 @@ func (s *System) rebuildProfileInterest(ctx context.Context, store userstore.Use break } } - if len(itemIDs) == 0 { - return nil + // History imports (watch providers, history import runs) may record + // watched episodes without any progress row; they are watch relationships + // all the same. + for offset := 0; ; offset += pageSize { + history, err := store.ListCompletedHistory(ctx, userstore.CompletedHistoryQuery{ + ProfileID: profileID, + Limit: pageSize, + Offset: offset, + }) + if err != nil { + return fmt.Errorf("list completed history: %w", err) + } + for _, entry := range history { + itemIDs[entry.MediaItemID] = struct{}{} + } + if len(history) < pageSize { + break + } } seriesIDs, err := s.batchResolveSeries(ctx, itemIDs) if err != nil { return err } + existing, err := s.Interests.ListSeriesForProfile(ctx, profileID) + if err != nil { + return err + } + for _, seriesID := range existing { + seriesIDs[seriesID] = struct{}{} + } for seriesID := range seriesIDs { if ctx.Err() != nil { return ctx.Err() diff --git a/internal/notifications/webhook_guard.go b/internal/notifications/webhook_guard.go index b5bbe8b5..4b826afc 100644 --- a/internal/notifications/webhook_guard.go +++ b/internal/notifications/webhook_guard.go @@ -76,7 +76,7 @@ func ValidateWebhookURL(rawURL string, allowPrivate bool) (host string, err erro if err != nil { return "", fmt.Errorf("invalid URL") } - if parsed.Scheme != "https" { + if parsed.Scheme != schemeHTTPS { return "", fmt.Errorf("webhook URLs must use https") } host = parsed.Hostname() diff --git a/internal/notifications/webhook_http.go b/internal/notifications/webhook_http.go index a35443e4..e7f6721a 100644 --- a/internal/notifications/webhook_http.go +++ b/internal/notifications/webhook_http.go @@ -70,7 +70,7 @@ func newWebhookHTTPClient(allowPrivate func() bool) *http.Client { if len(via) >= webhookMaxRedirects { return errors.New("too many redirects") } - if req.URL.Scheme != "https" { + if req.URL.Scheme != schemeHTTPS { return errors.New("redirect to non-https destination") } return nil @@ -100,6 +100,12 @@ func sendWebhook(ctx context.Context, client *http.Client, url string, body []by if err != nil { return result(false, 0, "invalid webhook URL") } + // HTTPS is enforced at registration; re-check at the last layer before + // the wire so delivery never depends on upstream validation staying + // perfect (webhook URLs are credentials and must not travel cleartext). + if req.URL.Scheme != schemeHTTPS { + return result(false, 0, "invalid webhook URL") + } req.Header.Set("Content-Type", "application/json") req.Header.Set("User-Agent", webhookUserAgent) for key, value := range headers { @@ -123,13 +129,35 @@ func sendWebhook(ctx context.Context, client *http.Client, url string, body []by out.Message = fmt.Sprintf("%d %s", resp.StatusCode, out.Message) } if resp.StatusCode == http.StatusTooManyRequests { - if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && seconds > 0 { - out.RetryAfter = time.Duration(seconds) * time.Second - } + out.RetryAfter = parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()) } return out } +// maxRetryAfter caps how far a destination's Retry-After header can push the +// next attempt; the longest scheduled backoff is 24h and a buggy or hostile +// header must not park deliveries beyond it. +const maxRetryAfter = 24 * time.Hour + +// parseRetryAfter interprets a Retry-After header in both RFC 9110 forms — +// delta-seconds and HTTP-date — returning 0 when absent or unusable. +func parseRetryAfter(header string, now time.Time) time.Duration { + header = strings.TrimSpace(header) + if header == "" { + return 0 + } + delay := time.Duration(0) + if seconds, err := strconv.Atoi(header); err == nil { + delay = time.Duration(seconds) * time.Second + } else if when, err := http.ParseTime(header); err == nil { + delay = when.Sub(now) + } + if delay <= 0 { + return 0 + } + return min(delay, maxRetryAfter) +} + // classifyWebhookError maps transport errors to short diagnostic classes. // Messages must stay free of URLs and payload contents. func classifyWebhookError(err error) string { diff --git a/internal/notifications/webhook_payload_discord.go b/internal/notifications/webhook_payload_discord.go index d1845df5..86090d0b 100644 --- a/internal/notifications/webhook_payload_discord.go +++ b/internal/notifications/webhook_payload_discord.go @@ -137,7 +137,7 @@ func discordEmbedTitle(row DeliveryRow) string { } series := row.SeriesTitle if series == "" { - series = "New episode" + series = genericEpisodeTitle } var code string if row.SeasonNumber != nil && row.EpisodeNumber != nil { diff --git a/internal/notifications/webhook_repo.go b/internal/notifications/webhook_repo.go index ef988489..95ed052c 100644 --- a/internal/notifications/webhook_repo.go +++ b/internal/notifications/webhook_repo.go @@ -8,6 +8,7 @@ import ( "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) @@ -121,22 +122,47 @@ func (r *WebhookRepository) getByIDUnscoped(ctx context.Context, id string) (*We return hook, nil } -// CountByProfile returns the profile's webhook count (for the per-profile cap). -func (r *WebhookRepository) CountByProfile(ctx context.Context, profileID string) (int, error) { - var count int - err := r.pool.QueryRow(ctx, - `SELECT count(*) FROM notification_webhooks WHERE profile_id = $1`, profileID, - ).Scan(&count) - return count, err -} - // ErrWebhookNameTaken is returned when a profile already has a webhook with // the requested name. var ErrWebhookNameTaken = errors.New("a webhook with this name already exists") -// Insert persists a new webhook. -func (r *WebhookRepository) Insert(ctx context.Context, hook Webhook) error { - _, err := r.pool.Exec(ctx, ` +// isWebhookNameViolation reports whether err is the unique violation on the +// per-profile webhook name constraint, via the typed pgx error (string +// matching on error text would break if message formatting changes). +func isWebhookNameViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && + pgErr.Code == "23505" && + pgErr.ConstraintName == "notification_webhooks_profile_name_key" +} + +// InsertWithLimit persists a new webhook unless the profile is already at +// maxPerProfile. The count and insert run under a per-profile advisory +// transaction lock so concurrent creates cannot both pass the check and push +// the profile past its cap. +func (r *WebhookRepository) InsertWithLimit(ctx context.Context, hook Webhook, maxPerProfile int) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin webhook insert: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + if _, err := tx.Exec(ctx, + `SELECT pg_advisory_xact_lock(hashtextextended('notification_webhooks:' || $1, 0))`, + hook.ProfileID); err != nil { + return fmt.Errorf("lock webhook quota: %w", err) + } + var count int + if err := tx.QueryRow(ctx, + `SELECT count(*) FROM notification_webhooks WHERE profile_id = $1`, hook.ProfileID, + ).Scan(&count); err != nil { + return fmt.Errorf("count webhooks: %w", err) + } + if count >= maxPerProfile { + return ErrWebhookLimit + } + + if _, err := tx.Exec(ctx, ` INSERT INTO notification_webhooks (id, user_id, profile_id, name, type, url_ciphertext, url_host, signing_secret_ciphertext, enabled, @@ -146,13 +172,15 @@ func (r *WebhookRepository) Insert(ctx context.Context, hook Webhook) error { hook.ID, hook.UserID, hook.ProfileID, hook.Name, hook.Type, hook.URLCiphertext, hook.URLHost, hook.SigningSecretCiphertext, hook.Enabled, hook.NotifyFavorites, hook.NotifyWatchlist, hook.NotifyContinueWatching, hook.NotifyNextUp, - hook.NotifyRequests) - if err != nil { - if strings.Contains(err.Error(), "notification_webhooks_profile_name_key") { + hook.NotifyRequests); err != nil { + if isWebhookNameViolation(err) { return ErrWebhookNameTaken } return fmt.Errorf("insert webhook: %w", err) } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit webhook insert: %w", err) + } return nil } @@ -175,7 +203,7 @@ func (r *WebhookRepository) Update(ctx context.Context, hook Webhook) error { hook.NotifyRequests, hook.ConsecutiveFailures, hook.DisabledReason) if err != nil { - if strings.Contains(err.Error(), "notification_webhooks_profile_name_key") { + if isWebhookNameViolation(err) { return ErrWebhookNameTaken } return fmt.Errorf("update webhook: %w", err) diff --git a/internal/notifications/webhook_sender.go b/internal/notifications/webhook_sender.go index 8a50288b..f8c12a52 100644 --- a/internal/notifications/webhook_sender.go +++ b/internal/notifications/webhook_sender.go @@ -162,7 +162,19 @@ func (s *webhookSender) processAttempt(ctx context.Context, attempt DeliveryAtte } row, err := s.deliveries.GetRowByID(ctx, attempt.NotificationDeliveryID) - if err != nil || row == nil { + if err != nil { + // Transient lookup failure: leave the claimed attempt alone so the + // lease expires and the retry worker reclaims it, instead of + // permanently failing the delivery over a database blip. + if ctx.Err() == nil { + s.logger.Warn("webhook delivery lookup failed", + "attempt_id", attempt.ID, + "delivery_id", attempt.NotificationDeliveryID, + "error", err) + } + return + } + if row == nil { _ = s.webhooks.FinalizeAttempt(ctx, attempt.ID, WebhookOutcomeFailed, attempt.AttemptNumber+1, nil, "delivery row missing", nil) return diff --git a/internal/notifications/webhook_service.go b/internal/notifications/webhook_service.go index dca26874..b87787cd 100644 --- a/internal/notifications/webhook_service.go +++ b/internal/notifications/webhook_service.go @@ -90,14 +90,6 @@ func (s *WebhookService) Create(ctx context.Context, userID int, profileID strin return nil, "", err } - count, err := s.repo.CountByProfile(ctx, profileID) - if err != nil { - return nil, "", err - } - if count >= s.settings.WebhooksMaxPerProfile(ctx) { - return nil, "", ErrWebhookLimit - } - rawURL := strings.TrimSpace(*input.URL) host, err := ValidateWebhookURL(rawURL, s.settings.WebhooksAllowPrivateDestinations(ctx)) if err != nil { @@ -108,16 +100,25 @@ func (s *WebhookService) Create(ctx context.Context, userID int, profileID strin if input.Type != nil { hookType = strings.TrimSpace(*input.Type) } + isDiscordURL := discordWebhookURL(rawURL) switch hookType { case "": hookType = WebhookTypeGeneric - if discordWebhookURL(rawURL) { + if isDiscordURL { hookType = WebhookTypeDiscord } case WebhookTypeDiscord, WebhookTypeGeneric: default: return nil, "", fmt.Errorf("%w: type must be discord or generic", ErrWebhookInvalid) } + // An explicit type must match the destination, or the sender would apply + // the wrong payload/signing behavior from the first delivery. + if hookType == WebhookTypeDiscord && !isDiscordURL { + return nil, "", fmt.Errorf("%w: type discord requires a Discord webhook URL", ErrWebhookInvalid) + } + if hookType == WebhookTypeGeneric && isDiscordURL { + return nil, "", fmt.Errorf("%w: Discord webhook URLs must use type discord", ErrWebhookInvalid) + } hook := Webhook{ ID: ulid.Make().String(), @@ -151,7 +152,9 @@ func (s *WebhookService) Create(ctx context.Context, userID int, profileID strin hook.SigningSecretCiphertext = &ciphertext } - if err := s.repo.Insert(ctx, hook); err != nil { + // The per-profile cap is enforced inside the insert (advisory-locked + // count + insert), so concurrent creates cannot both slip past it. + if err := s.repo.InsertWithLimit(ctx, hook, s.settings.WebhooksMaxPerProfile(ctx)); err != nil { if errors.Is(err, ErrWebhookNameTaken) { return nil, "", fmt.Errorf("%w: %s", ErrWebhookInvalid, err.Error()) } diff --git a/internal/notifications/webhook_types.go b/internal/notifications/webhook_types.go index 9ab71d13..7cfe876a 100644 --- a/internal/notifications/webhook_types.go +++ b/internal/notifications/webhook_types.go @@ -30,6 +30,13 @@ const DeliveryTypeWebhookAutoDisabled = "webhook.auto_disabled" // build does not know (the type registry is extensible). const genericNotificationTitle = "Silo notification" +// genericEpisodeTitle is the display fallback for episode.available rows +// whose catalog metadata is missing or was deleted. +const genericEpisodeTitle = "New episode" + +// schemeHTTPS is the only scheme outbound notification HTTP traffic may use. +const schemeHTTPS = "https" + // Webhook is a profile-scoped outbound destination. URL and signing secret // are stored as enc:v1: envelopes and never leave the server. type Webhook struct { diff --git a/internal/notifications/webpush_repo.go b/internal/notifications/webpush_repo.go index ed050412..e0319b6b 100644 --- a/internal/notifications/webpush_repo.go +++ b/internal/notifications/webpush_repo.go @@ -76,7 +76,27 @@ func scanWebPushSubscriptions(rows pgx.Rows) ([]WebPushSubscription, error) { // caller's (user, profile): one browser endpoint notifies exactly one // profile, the one that subscribed most recently. func (r *WebPushRepository) Upsert(ctx context.Context, sub WebPushSubscription) (*WebPushSubscription, error) { - row := r.pool.QueryRow(ctx, ` + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, fmt.Errorf("begin web push upsert: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + // Reassignment keeps the row id, so undelivered attempts enqueued for the + // previous owner would otherwise be sent to the new owner's browser once + // the retry worker reclaims them. Cancel them before the ownership flips. + if _, err := tx.Exec(ctx, ` + DELETE FROM web_push_delivery_attempts + WHERE outcome IN ('pending', 'retrying') + AND subscription_id IN ( + SELECT id FROM web_push_subscriptions + WHERE endpoint = $1 AND profile_id <> $2 + )`, + sub.Endpoint, sub.ProfileID); err != nil { + return nil, fmt.Errorf("purge reassigned web push attempts: %w", err) + } + + row := tx.QueryRow(ctx, ` INSERT INTO web_push_subscriptions (id, user_id, profile_id, endpoint, p256dh, auth, device_name) VALUES ($1, $2, $3, $4, $5, $6, $7) @@ -95,6 +115,9 @@ func (r *WebPushRepository) Upsert(ctx context.Context, sub WebPushSubscription) if err != nil { return nil, fmt.Errorf("upsert web push subscription: %w", err) } + if err := tx.Commit(ctx); err != nil { + return nil, fmt.Errorf("commit web push upsert: %w", err) + } return saved, nil } diff --git a/internal/notifications/webpush_sender.go b/internal/notifications/webpush_sender.go index 118bf4df..29044173 100644 --- a/internal/notifications/webpush_sender.go +++ b/internal/notifications/webpush_sender.go @@ -7,7 +7,6 @@ import ( "io" "log/slog" "net/http" - "strconv" "sync" "time" @@ -152,11 +151,29 @@ func (s *webPushSender) processAttempt(ctx context.Context, attempt DeliveryAtte return } row, err := s.deliveries.GetRowByID(ctx, attempt.NotificationDeliveryID) - if err != nil || row == nil { + if err != nil { + // Transient lookup failure: let the claim lease expire and the retry + // worker reclaim, instead of permanently failing the delivery. + if ctx.Err() == nil { + s.logger.Warn("web push delivery lookup failed", + "attempt_id", attempt.ID, + "delivery_id", attempt.NotificationDeliveryID, + "error", err) + } + return + } + if row == nil { _ = s.subscriptions.FinalizeAttempt(ctx, attempt.ID, WebhookOutcomeFailed, attempt.AttemptNumber+1, nil, "delivery row missing", nil) return } + if row.ProfileID != sub.ProfileID { + // The endpoint was reassigned to a different profile between enqueue + // and dispatch; this delivery belongs to the previous owner. + _ = s.subscriptions.FinalizeAttempt(ctx, attempt.ID, WebhookOutcomeFailed, + attempt.AttemptNumber+1, nil, "subscription reassigned", nil) + return + } publicKey, privateKey, err := s.service.vapidKeys(ctx) if err != nil { @@ -241,9 +258,7 @@ func (s *webPushSender) send(ctx context.Context, sub *WebPushSubscription, mess defer func() { _ = resp.Body.Close() }() _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 16<<10)) if resp.StatusCode == http.StatusTooManyRequests { - if seconds, parseErr := strconv.Atoi(resp.Header.Get("Retry-After")); parseErr == nil && seconds > 0 { - retryAfter = time.Duration(seconds) * time.Second - } + retryAfter = parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()) } return resp.StatusCode, retryAfter, nil } diff --git a/internal/notifications/webpush_service.go b/internal/notifications/webpush_service.go index 5ec0fcea..57805a46 100644 --- a/internal/notifications/webpush_service.go +++ b/internal/notifications/webpush_service.go @@ -13,10 +13,11 @@ import ( ) // VAPID key settings. The keypair is stored as a single JSON value (encrypted -// at rest via SensitiveSettingKeys) so both halves persist atomically: two -// concurrent provisioners can each lose the race wholesale but can never -// strand a mismatched public/private pair. Clients receive the public half -// from the capability endpoint, never from the settings store directly. +// at rest via SensitiveSettingKeys) so both halves persist atomically, and it +// is written with SetIfAbsent so exactly one concurrent provisioner's pair can +// ever land — a mismatched public/private pair or split-brain identity across +// nodes is impossible. Clients receive the public half from the capability +// endpoint, never from the settings store directly. const ( SettingWebPushEnabled = "notifications.web_push_enabled" SettingWebPushVAPIDKeypair = "notifications.web_push.vapid_keypair" //nolint:gosec // setting key name, not a credential @@ -37,6 +38,10 @@ func (s *Settings) WebPushEnabled(ctx context.Context) bool { // catalog.EncryptedSettingsRepo, which encrypts sensitive keys on write. type SettingWriter interface { Set(ctx context.Context, key, value string) error + // SetIfAbsent writes only when the key has no value yet, reporting whether + // this writer won. Generated credentials must be provisioned single-writer: + // concurrent nodes may race to generate, but exactly one value can land. + SetIfAbsent(ctx context.Context, key, value string) (bool, error) } // ErrWebPushInvalid marks rejected subscription input. @@ -70,8 +75,13 @@ func (s *WebPushService) vapidKeys(ctx context.Context) (publicKey, privateKey s return s.publicKey, s.privateKey, nil } - stored := s.loadKeypair(ctx) - if stored.Public == "" || stored.Private == "" { + stored, found, err := s.loadKeypair(ctx) + if err != nil { + // Never reprovision over a read or decode failure: rotating the VAPID + // identity silently invalidates every existing browser subscription. + return "", "", err + } + if !found { if s.writer == nil { return "", "", errors.New("web push requires a writable settings store") } @@ -83,15 +93,24 @@ func (s *WebPushService) vapidKeys(ctx context.Context) (publicKey, privateKey s if marshalErr != nil { return "", "", fmt.Errorf("encode VAPID keypair: %w", marshalErr) } - if err := s.writer.Set(ctx, SettingWebPushVAPIDKeypair, string(data)); err != nil { - return "", "", fmt.Errorf("persist VAPID keypair: %w", err) + // Conditional write: with concurrent provisioners exactly one generated + // pair can ever land, so no node can cache a pair another node's write + // later overwrites (split-brain VAPID identities). + won, setErr := s.writer.SetIfAbsent(ctx, SettingWebPushVAPIDKeypair, string(data)) + if setErr != nil { + return "", "", fmt.Errorf("persist VAPID keypair: %w", setErr) } - // Re-read and adopt whatever the store now holds: if another process - // provisioned concurrently, the last write won and every node must - // converge on that pair rather than caching its own loser. - stored = s.loadKeypair(ctx) - if stored.Public == "" || stored.Private == "" { + if won { stored = vapidKeypair{Public: public, Private: private} + } else { + // Another node provisioned first: adopt its pair. + stored, found, err = s.loadKeypair(ctx) + if err != nil { + return "", "", err + } + if !found { + return "", "", errors.New("VAPID keypair disappeared during provisioning") + } } } s.publicKey = stored.Public @@ -101,21 +120,27 @@ func (s *WebPushService) vapidKeys(ctx context.Context) (publicKey, privateKey s // loadKeypair reads the persisted keypair directly from the settings reader, // bypassing the Settings facade cache: provisioning must observe the latest -// stored value, not a seconds-old cached miss. A corrupt value reads as -// unprovisioned and is overwritten with a fresh pair. -func (s *WebPushService) loadKeypair(ctx context.Context) vapidKeypair { - var keys vapidKeypair +// stored value, not a seconds-old cached miss. found is true only for a +// complete stored pair; read and decode failures surface as errors so callers +// never mistake them for "not provisioned yet". +func (s *WebPushService) loadKeypair(ctx context.Context) (keys vapidKeypair, found bool, err error) { if s.settings == nil || s.settings.reader == nil { - return keys + return vapidKeypair{}, false, nil } raw, err := s.settings.reader.Get(ctx, SettingWebPushVAPIDKeypair) - if err != nil || strings.TrimSpace(raw) == "" { - return vapidKeypair{} + if err != nil { + return vapidKeypair{}, false, fmt.Errorf("read VAPID keypair: %w", err) + } + if strings.TrimSpace(raw) == "" { + return vapidKeypair{}, false, nil } if err := json.Unmarshal([]byte(raw), &keys); err != nil { - return vapidKeypair{} + return vapidKeypair{}, false, fmt.Errorf("decode stored VAPID keypair: %w", err) } - return keys + if keys.Public == "" || keys.Private == "" { + return vapidKeypair{}, false, errors.New("stored VAPID keypair is incomplete") + } + return keys, true, nil } // PublicKey returns the VAPID application server key clients subscribe with. diff --git a/migrations/sql/20260611202951_episode_ordinal_checks.sql b/migrations/sql/20260611202951_episode_ordinal_checks.sql new file mode 100644 index 00000000..46f31fb3 --- /dev/null +++ b/migrations/sql/20260611202951_episode_ordinal_checks.sql @@ -0,0 +1,33 @@ +-- +goose Up +-- +goose StatementBegin +-- Enforce the episode-key invariants (episode_key.go: EpisodeKey / +-- ValidEpisodeOrdinals) at the database layer too, so direct SQL or future +-- backfill paths cannot insert rows that break the collision-free ordering +-- contract the fanout and suppression logic depend on. Bounds mirror the Go +-- constants: season <= 2146 keeps season * 1,000,000 + episode inside int4. +ALTER TABLE public.episode_availability + ADD CONSTRAINT episode_availability_ordinals_check CHECK ( + season_number >= 0 + AND season_number <= 2146 + AND episode_number >= 0 + AND episode_number < 1000000 + AND episode_key = season_number * 1000000 + episode_number + ); + +ALTER TABLE public.release_events + ADD CONSTRAINT release_events_ordinals_check CHECK ( + season_number >= 0 + AND season_number <= 2146 + AND episode_number >= 0 + AND episode_number < 1000000 + AND episode_key = season_number * 1000000 + episode_number + ); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE public.episode_availability + DROP CONSTRAINT IF EXISTS episode_availability_ordinals_check; +ALTER TABLE public.release_events + DROP CONSTRAINT IF EXISTS release_events_ordinals_check; +-- +goose StatementEnd diff --git a/web/public/sw.js b/web/public/sw.js index a20dda06..e20a4489 100644 --- a/web/public/sw.js +++ b/web/public/sw.js @@ -32,7 +32,18 @@ self.addEventListener("push", (event) => { self.addEventListener("notificationclick", (event) => { event.notification.close(); - const url = (event.notification.data && event.notification.data.url) || "/notifications"; + // Notifications navigate same-origin only: payload data is server-built, + // but a notification surface must never become an open redirect. + const rawUrl = (event.notification.data && event.notification.data.url) || "/notifications"; + let url = "/notifications"; + try { + const parsed = new URL(rawUrl, self.location.origin); + if (parsed.origin === self.location.origin) { + url = `${parsed.pathname}${parsed.search}${parsed.hash}`; + } + } catch { + // keep the safe default + } event.waitUntil( self.clients.matchAll({ type: "window", includeUncontrolled: true }).then((clientList) => { for (const client of clientList) { diff --git a/web/src/components/RealtimeEventsProvider.tsx b/web/src/components/RealtimeEventsProvider.tsx index 1f11b95b..a75d5d9e 100644 --- a/web/src/components/RealtimeEventsProvider.tsx +++ b/web/src/components/RealtimeEventsProvider.tsx @@ -739,10 +739,15 @@ export function RealtimeEventsProvider({ children }: { children: ReactNode }) { (entry) => entry.channel === "notifications" && entry.code === "profile_required", ); if (profileRequired && activeProfileIDRef.current) { + // Rebinding requires a fresh handshake (tickets are consumed at + // upgrade time), so the shared socket must close. The backoff + // grows to 5 minutes so a persistent notifications-only outage + // costs the catalog/user_state channels one brief flap per + // cycle instead of a permanent fast reconnect loop. profileRebindAttemptsRef.current += 1; nextReconnectDelayRef.current = Math.min( - 30_000, - 1_000 * 2 ** Math.min(profileRebindAttemptsRef.current, 5), + 300_000, + 1_000 * 2 ** Math.min(profileRebindAttemptsRef.current, 9), ); socket.close(); } else { diff --git a/web/src/pages/Notifications.tsx b/web/src/pages/Notifications.tsx index 84971e88..efc95b4d 100644 --- a/web/src/pages/Notifications.tsx +++ b/web/src/pages/Notifications.tsx @@ -203,7 +203,7 @@ function NotificationRow({ } function NotificationPreferencesPopover() { - const { data: prefs, isLoading } = useNotificationPreferences(); + const { data: prefs, isLoading, refetch } = useNotificationPreferences(); const updatePrefs = useUpdateNotificationPreferences(); const toggles: Array<{ @@ -252,12 +252,19 @@ function NotificationPreferencesPopover() { - {isLoading || !prefs ? ( + {isLoading ? (
+ ) : !prefs ? ( +
+

Couldn’t load preferences.

+ +
) : (
{toggles.map((toggle, index) => (