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 <noreply@anthropic.com>

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)
 	}

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)

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) {

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`)

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)

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 {

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

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.

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 /

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()

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()

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 {

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 {

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)

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

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())
 		}

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 {

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
 }
 

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
 }

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.

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

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) {

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 {

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() {
         </Button>
       </PopoverTrigger>
       <PopoverContent align="end" className="w-80">
-        {isLoading || !prefs ? (
+        {isLoading ? (
           <div className="space-y-3">
             <Skeleton className="h-6 w-full" />
             <Skeleton className="h-6 w-full" />
             <Skeleton className="h-6 w-full" />
           </div>
+        ) : !prefs ? (
+          <div className="space-y-3">
+            <p className="text-muted-foreground text-sm">Couldn’t load preferences.</p>
+            <Button size="sm" variant="outline" onClick={() => void refetch()}>
+              Retry
+            </Button>
+          </div>
         ) : (
           <div className="space-y-4">
             {toggles.map((toggle, index) => (