feat(notifications): request-fulfilled notifications across all channels

Notify the requesting profile once its media request is actually present
in the catalog (roadmap 06, item 2). Completion transitions stay
notification-agnostic; a presence-gated pass at the end of each
reconcile run fires the notice, so it means "watchable in Silo", not
"download finished".

- New System.DispatchOperational: delivery insert + webhook/web-push
  outbox enqueue in one transaction, post-commit multi-dispatch. The
  webhook auto-disable notice now rides the same path (replacing its
  hand-rolled hub publish and the now-removed InsertOperational), which
  also delivers auto-disable notices over web push.
- At-most-once delivery: partial unique index on
  (profile_id, reason_flags->>'request_id') plus a fulfilled_notified_at
  marker on media_requests, backfilled for pre-existing completed
  requests so deploys never flood.
- Per-webhook notify_requests toggle (default on) through repo, service,
  API, and settings UI; gated independently of the episode reason flags.
- request.fulfilled rendering in web inbox, realtime toast, web push
  payload, and Discord/generic webhook payloads, deep-linking to the
  matched catalog item.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Quick
2026-06-11 16:04:22 -04:00
co-authored by Claude Fable 5
parent 4fd21c7aee
commit d9e27da59e
26 changed files with 728 additions and 54 deletions
+3
View File
@@ -1663,6 +1663,9 @@ func main() {
)
requestReconcileSvc.SetEntitlementResolver(mediarequests.NewAccessEntitlements(reconcileResolver))
}
if notificationSystem != nil {
requestReconcileSvc.SetFulfillmentNotifier(notifications.NewRequestFulfillmentNotifier(notificationSystem))
}
taskMgr.Register(tasks.NewReconcileRequestsTask(requestReconcileSvc, 100))
if deps.FolderRepo != nil && deps.LibraryScanQueue != nil && pluginService != nil && pluginInstallationStore != nil {
autoscanRepo := autoscan.NewRepository(deps.DB, deps.SecretCipher)
@@ -24,6 +24,7 @@ type webhookResponse struct {
NotifyWatchlist bool `json:"notify_watchlist"`
NotifyContinueWatching bool `json:"notify_continue_watching"`
NotifyNextUp bool `json:"notify_next_up"`
NotifyRequests bool `json:"notify_requests"`
ConsecutiveFailures int `json:"consecutive_failures"`
DisabledReason *string `json:"disabled_reason"`
LastSuccessAt *time.Time `json:"last_success_at"`
@@ -45,6 +46,7 @@ func webhookToResponse(hook notifications.Webhook) webhookResponse {
NotifyWatchlist: hook.NotifyWatchlist,
NotifyContinueWatching: hook.NotifyContinueWatching,
NotifyNextUp: hook.NotifyNextUp,
NotifyRequests: hook.NotifyRequests,
ConsecutiveFailures: hook.ConsecutiveFailures,
DisabledReason: hook.DisabledReason,
LastSuccessAt: hook.LastSuccessAt,
@@ -63,6 +65,7 @@ type webhookRequest struct {
NotifyWatchlist *bool `json:"notify_watchlist"`
NotifyContinueWatching *bool `json:"notify_continue_watching"`
NotifyNextUp *bool `json:"notify_next_up"`
NotifyRequests *bool `json:"notify_requests"`
}
func (r webhookRequest) toInput() notifications.WebhookInput {
@@ -75,6 +78,7 @@ func (r webhookRequest) toInput() notifications.WebhookInput {
NotifyWatchlist: r.NotifyWatchlist,
NotifyContinueWatching: r.NotifyContinueWatching,
NotifyNextUp: r.NotifyNextUp,
NotifyRequests: r.NotifyRequests,
}
}
-22
View File
@@ -228,28 +228,6 @@ func (r *DeliveryRepository) GetRowByID(ctx context.Context, id string) (*Delive
return &out[0], nil
}
// InsertOperational inserts a single operational delivery (e.g.
// webhook.auto_disabled) outside the fanout path. Returns nil when the row
// deduped away.
func (r *DeliveryRepository) InsertOperational(ctx context.Context, delivery Delivery) (*InsertedDelivery, error) {
tx, err := r.pool.Begin(ctx)
if err != nil {
return nil, fmt.Errorf("begin operational delivery tx: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
inserted, err := r.BulkInsert(ctx, tx, []Delivery{delivery})
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("commit operational delivery: %w", err)
}
if len(inserted) == 0 {
return nil, nil
}
return &inserted[0], nil
}
// RecentUnread returns the newest unread rows for the websocket snapshot.
func (r *DeliveryRepository) RecentUnread(ctx context.Context, profileID string, limit int) ([]DeliveryRow, error) {
rows, err := r.pool.Query(ctx,
@@ -0,0 +1,99 @@
package notifications
import (
"context"
"fmt"
"github.com/oklog/ulid/v2"
)
// OperationalDispatch describes how one operational delivery (a non-fanout
// notice such as webhook.auto_disabled or request.fulfilled) reaches the
// per-target channels. WebhookFilter selects which of the profile's enabled
// webhooks receive it; nil means the type must not reach webhooks at all
// (e.g. the auto-disable notice's loop guard). Web push has no per-type
// filter: profile-level gating happens before dispatch.
type OperationalDispatch struct {
WebhookFilter func(Webhook) bool
}
// DispatchOperational durably creates one operational delivery. The inbox row
// and the per-target webhook / web push outbox rows commit in a single
// transaction — a crash afterwards delays channel sends instead of dropping
// them, because the retry workers recover pending outbox rows — then realtime
// and channel dispatch run post-commit. Returns nil when the delivery deduped
// away (the partial unique indexes make operational notices idempotent).
func (s *System) DispatchOperational(ctx context.Context, delivery Delivery, opts OperationalDispatch) (*InsertedDelivery, error) {
if s == nil {
return nil, nil
}
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, fmt.Errorf("begin operational dispatch tx: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
inserted, err := s.Deliveries.BulkInsert(ctx, tx, []Delivery{delivery})
if err != nil {
return nil, err
}
if len(inserted) == 0 {
return nil, nil
}
row := inserted[0]
if opts.WebhookFilter != nil && s.webhookRepo != nil && s.Settings.WebhooksEnabled(ctx) {
hooksByProfile, err := s.webhookRepo.ListEnabledByProfiles(ctx, tx, []string{delivery.ProfileID})
if err != nil {
return nil, err
}
attempts := make([]DeliveryAttempt, 0, 2)
for _, hook := range hooksByProfile[delivery.ProfileID] {
if !opts.WebhookFilter(hook) {
continue
}
attempts = append(attempts, DeliveryAttempt{
ID: ulid.Make().String(),
NotificationDeliveryID: row.ID,
TargetID: hook.ID,
})
}
if err := s.webhookRepo.EnqueueAttempts(ctx, tx, attempts); err != nil {
return nil, err
}
}
if s.webPushRepo != nil && s.Settings.WebPushEnabled(ctx) {
subsByProfile, err := s.webPushRepo.ListEnabledByProfiles(ctx, tx, []string{delivery.ProfileID})
if err != nil {
return nil, err
}
attempts := make([]DeliveryAttempt, 0, 2)
for _, sub := range subsByProfile[delivery.ProfileID] {
attempts = append(attempts, DeliveryAttempt{
ID: ulid.Make().String(),
NotificationDeliveryID: row.ID,
TargetID: sub.ID,
})
}
if err := s.webPushRepo.EnqueueAttempts(ctx, tx, attempts); err != nil {
return nil, err
}
}
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("commit operational dispatch: %w", err)
}
// Post-commit dispatch is best-effort: the durable inbox row covers
// websocket reconnect, and the retry workers recover the outbox rows.
full, err := s.Deliveries.GetRowByID(ctx, row.ID)
if err != nil || full == nil {
s.logger.Warn("operational delivery reload failed",
"delivery_id", row.ID, "error", err)
return &row, nil
}
if err := s.dispatcher.Dispatch(ctx, *full); err != nil {
s.logger.Warn("operational delivery dispatch failed",
"delivery_id", row.ID, "error", err)
}
return &row, nil
}
@@ -0,0 +1,90 @@
package notifications
import (
"context"
"encoding/json"
"fmt"
"github.com/Silo-Server/silo-server/internal/requests"
"github.com/oklog/ulid/v2"
)
// DeliveryTypeRequestFulfilled is the operational notice posted to the
// requesting profile once its media request is present in the catalog
// (docs/superpowers/plans/notifications/06, item 2). Its reason_flags carry
// {"request_id","tmdb_id","media_type"} instead of reason booleans; a partial
// unique index on (profile_id, request_id) makes the insert idempotent, and
// the per-webhook notify_requests flag gates the webhook channel.
const DeliveryTypeRequestFulfilled = "request.fulfilled"
// RequestFulfilledFlags is the decoded reason_flags shape for
// request.fulfilled deliveries.
type RequestFulfilledFlags struct {
RequestID string `json:"request_id"`
TMDBID int `json:"tmdb_id"`
MediaType string `json:"media_type"`
}
// parseRequestFulfilledFlags decodes a request.fulfilled delivery's
// reason_flags; other types decode to the zero value.
func parseRequestFulfilledFlags(raw []byte) RequestFulfilledFlags {
var flags RequestFulfilledFlags
if len(raw) > 0 {
_ = json.Unmarshal(raw, &flags)
}
return flags
}
// RequestFulfillmentNotifier adapts the notification system to
// requests.FulfillmentNotifier: it gates on the profile's master toggle and
// dispatches one durable request.fulfilled delivery across all channels.
type RequestFulfillmentNotifier struct {
system *System
}
// NewRequestFulfillmentNotifier creates the adapter.
func NewRequestFulfillmentNotifier(system *System) *RequestFulfillmentNotifier {
return &RequestFulfillmentNotifier{system: system}
}
// NotifyFulfilled implements requests.FulfillmentNotifier. contentID is the
// matched catalog item: deliveryRowSelect joins media_items on series_id, so
// that one field renders the title, poster, and deep link for movies and
// series alike. Returning nil without dispatching (master toggle off, missing
// attribution) still counts as handled — the caller stamps the request either
// way.
func (n *RequestFulfillmentNotifier) NotifyFulfilled(ctx context.Context, req requests.Request, contentID string) error {
if n == nil || n.system == nil {
return nil
}
if req.RequestedByProfileID == "" || req.RequestedByUserID <= 0 {
return nil // legacy rows without attribution have no recipient
}
prefs, err := n.system.Preferences.Get(ctx, req.RequestedByProfileID)
if err != nil {
return err
}
if !prefs.Enabled {
return nil
}
flags, err := json.Marshal(RequestFulfilledFlags{
RequestID: req.ID,
TMDBID: req.TMDBID,
MediaType: string(req.MediaType),
})
if err != nil {
return fmt.Errorf("marshal request fulfilled flags: %w", err)
}
delivery := Delivery{
ID: ulid.Make().String(),
UserID: req.RequestedByUserID,
ProfileID: req.RequestedByProfileID,
SeriesID: &contentID,
Type: DeliveryTypeRequestFulfilled,
ReasonFlags: flags,
}
_, err = n.system.DispatchOperational(ctx, delivery, OperationalDispatch{
WebhookFilter: func(hook Webhook) bool { return hook.NotifyRequests },
})
return err
}
+9 -3
View File
@@ -55,6 +55,10 @@ type System struct {
webhookRetry *WebhookRetryWorker
webPushRepo *WebPushRepository
webPushDispatcher *WebPushDispatcher
// dispatcher is the same MultiDispatcher the fanout worker uses; the
// operational dispatch path shares it so every delivery reaches every
// configured channel the same way.
dispatcher Dispatcher
pool *pgxpool.Pool
stores userstore.UserStoreProvider
@@ -94,7 +98,7 @@ func NewSystem(
var sender *webhookSender
if cipher != nil {
webhookRepo = NewWebhookRepository(pool)
sender = newWebhookSender(webhookRepo, deliveries, cipher, settings, hub)
sender = newWebhookSender(webhookRepo, deliveries, cipher, settings)
webhookService = newWebhookService(webhookRepo, cipher, settings, sender)
webhookDispatcher = newWebhookDispatcher(sender)
webhookRetry = newWebhookRetryWorker(sender)
@@ -116,7 +120,8 @@ func NewSystem(
dispatchers = append(dispatchers, webPushDispatcher)
}
fanout := NewFanoutWorker(pool, releases, interests, deliveries, preferences, settings, NewMultiDispatcher(dispatchers...))
multiDispatcher := NewMultiDispatcher(dispatchers...)
fanout := NewFanoutWorker(pool, releases, interests, deliveries, preferences, settings, multiDispatcher)
if webhookRepo != nil {
fanout.SetWebhookOutbox(webhookRepo, newProfileRateLimiter())
}
@@ -144,6 +149,7 @@ func NewSystem(
webhookRetry: webhookRetry,
webPushRepo: webPushRepo,
webPushDispatcher: webPushDispatcher,
dispatcher: multiDispatcher,
pool: pool,
stores: stores,
users: users,
@@ -151,7 +157,7 @@ func NewSystem(
}
wsDispatcher.payload = system.PayloadForRow
if sender != nil {
sender.payload = system.PayloadForRow
sender.operational = system.DispatchOperational
}
if webPushSenderInst != nil {
webPushSenderInst.payload = system.PayloadForRow
@@ -163,6 +163,85 @@ func TestBuildDiscordWebhookPayload(t *testing.T) {
}
}
func requestFulfilledTestRow() DeliveryRow {
contentID := "movie-123"
return DeliveryRow{
Delivery: Delivery{
ID: "01REQUEST",
ProfileID: "profile-1",
SeriesID: &contentID,
Type: DeliveryTypeRequestFulfilled,
ReasonFlags: []byte(`{"request_id":"01REQ","tmdb_id":438631,"media_type":"movie"}`),
CreatedAt: time.Date(2026, 6, 11, 12, 0, 0, 0, time.UTC),
},
SeriesTitle: "Dune",
}
}
func TestBuildDiscordWebhookPayloadRequestFulfilled(t *testing.T) {
payload, err := BuildDiscordWebhookPayload(requestFulfilledTestRow(), false)
if err != nil {
t.Fatalf("build failed: %v", err)
}
var body struct {
Embeds []struct {
Title string `json:"title"`
Description string `json:"description"`
Fields []struct {
Name string `json:"name"`
Value string `json:"value"`
} `json:"fields"`
} `json:"embeds"`
}
if err := json.Unmarshal(payload, &body); err != nil {
t.Fatalf("payload is not valid JSON: %v", err)
}
if len(body.Embeds) != 1 {
t.Fatalf("unexpected body shape: %+v", body)
}
embed := body.Embeds[0]
if embed.Title != "Dune" {
t.Fatalf("unexpected title %q", embed.Title)
}
if embed.Description != "Your media request is now available on Silo" {
t.Fatalf("unexpected description %q", embed.Description)
}
if len(embed.Fields) != 1 || embed.Fields[0].Name != "Type" || embed.Fields[0].Value != "Movie" {
t.Fatalf("expected a single Type=Movie field, got %+v", embed.Fields)
}
}
func TestGenericWebhookPayloadRequestFulfilled(t *testing.T) {
payload, err := BuildGenericWebhookPayload(requestFulfilledTestRow(), "hook-1", false)
if err != nil {
t.Fatalf("build failed: %v", err)
}
var body struct {
Type string `json:"type"`
Series *struct {
ID string `json:"id"`
Title string `json:"title"`
} `json:"series"`
Request *struct {
ID string `json:"id"`
TMDBID int `json:"tmdb_id"`
MediaType string `json:"media_type"`
} `json:"request"`
}
if err := json.Unmarshal(payload, &body); err != nil {
t.Fatalf("payload is not valid JSON: %v", err)
}
if body.Type != DeliveryTypeRequestFulfilled {
t.Fatalf("unexpected type %q", body.Type)
}
if body.Request == nil || body.Request.ID != "01REQ" || body.Request.TMDBID != 438631 || body.Request.MediaType != "movie" {
t.Fatalf("unexpected request block: %+v", body.Request)
}
if body.Series == nil || body.Series.ID != "movie-123" || body.Series.Title != "Dune" {
t.Fatalf("unexpected series block: %+v", body.Series)
}
}
func TestBuildDiscordWebhookPayloadTestMarker(t *testing.T) {
payload, err := BuildDiscordWebhookPayload(webhookTestRow(), true)
if err != nil {
@@ -66,6 +66,9 @@ func BuildDiscordWebhookPayload(row DeliveryRow, test bool) ([]byte, error) {
title = truncateWithEllipsis(title, discordTitleLimit)
description := "New episode available on Silo"
if row.Type == DeliveryTypeRequestFulfilled {
description = "Your media request is now available on Silo"
}
footerText := "Silo"
if row.SeriesTitle != "" {
footerText = "Silo • " + truncateWithEllipsis(row.SeriesTitle, discordFooterLimit-16)
@@ -82,6 +85,11 @@ func BuildDiscordWebhookPayload(row DeliveryRow, test bool) ([]byte, error) {
Inline: true,
})
}
if row.Type == DeliveryTypeRequestFulfilled {
if mediaType := requestMediaTypeLabel(row.ReasonFlags); mediaType != "" {
fields = append(fields, discordEmbedField{Name: "Type", Value: mediaType, Inline: true})
}
}
if row.SeasonNumber != nil {
fields = append(fields, discordEmbedField{
Name: "Season",
@@ -116,7 +124,15 @@ func BuildDiscordWebhookPayload(row DeliveryRow, test bool) ([]byte, error) {
}
func discordEmbedTitle(row DeliveryRow) string {
if row.Type != DeliveryTypeEpisodeAvailable {
switch row.Type {
case DeliveryTypeRequestFulfilled:
if row.SeriesTitle != "" {
return row.SeriesTitle
}
return "Request fulfilled"
case DeliveryTypeEpisodeAvailable:
// Falls out of the switch into the episode title assembly below.
default:
return genericNotificationTitle
}
series := row.SeriesTitle
@@ -137,6 +153,19 @@ func discordEmbedTitle(row DeliveryRow) string {
}
}
// requestMediaTypeLabel renders a request.fulfilled delivery's media type as
// a display label; unknown values render nothing.
func requestMediaTypeLabel(reasonFlags []byte) string {
switch parseRequestFulfilledFlags(reasonFlags).MediaType {
case "movie":
return "Movie"
case "series":
return "Series"
default:
return ""
}
}
func discordEmbedColor(flags ReasonFlags) int {
switch {
case flags.Favorite:
@@ -28,6 +28,9 @@ type genericWebhookBody struct {
Reasons ReasonFlags `json:"reason_flags"`
Series *genericWebhookSeries `json:"series,omitempty"`
Episode *genericWebhookEpisode `json:"episode,omitempty"`
// Request is present for request.fulfilled deliveries; the catalog item is
// in Series (movies included — the field carries the matched item).
Request *genericWebhookRequest `json:"request,omitempty"`
}
type genericWebhookSeries struct {
@@ -42,6 +45,12 @@ type genericWebhookEpisode struct {
EpisodeNumber *int `json:"episode_number,omitempty"`
}
type genericWebhookRequest struct {
ID string `json:"id"`
TMDBID int `json:"tmdb_id,omitempty"`
MediaType string `json:"media_type,omitempty"`
}
// BuildGenericWebhookPayload renders a delivery as canonical Silo JSON. Pure
// function.
func BuildGenericWebhookPayload(row DeliveryRow, webhookID string, test bool) ([]byte, error) {
@@ -72,6 +81,14 @@ func BuildGenericWebhookPayload(row DeliveryRow, webhookID string, test bool) ([
EpisodeNumber: row.EpisodeNumber,
}
}
if row.Type == DeliveryTypeRequestFulfilled {
flags := parseRequestFulfilledFlags(row.ReasonFlags)
body.Request = &genericWebhookRequest{
ID: flags.RequestID,
TMDBID: flags.TMDBID,
MediaType: flags.MediaType,
}
}
return json.Marshal(body)
}
+10 -4
View File
@@ -25,6 +25,7 @@ const webhookColumns = `
id, user_id, profile_id, name, type, url_ciphertext, url_host,
signing_secret_ciphertext, enabled,
notify_favorites, notify_watchlist, notify_continue_watching, notify_next_up,
notify_requests,
consecutive_failures, disabled_reason,
last_success_at, last_failure_at, last_failure_status, last_failure_message,
created_at, updated_at`
@@ -35,6 +36,7 @@ func scanWebhook(row pgx.Row) (*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,
&hook.ConsecutiveFailures, &hook.DisabledReason,
&hook.LastSuccessAt, &hook.LastFailureAt, &hook.LastFailureStatus, &hook.LastFailureMessage,
&hook.CreatedAt, &hook.UpdatedAt,
@@ -138,11 +140,13 @@ func (r *WebhookRepository) Insert(ctx context.Context, hook Webhook) error {
INSERT INTO notification_webhooks
(id, user_id, profile_id, name, type, url_ciphertext, url_host,
signing_secret_ciphertext, enabled,
notify_favorites, notify_watchlist, notify_continue_watching, notify_next_up)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)`,
notify_favorites, notify_watchlist, notify_continue_watching, notify_next_up,
notify_requests)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`,
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.NotifyFavorites, hook.NotifyWatchlist, hook.NotifyContinueWatching, hook.NotifyNextUp,
hook.NotifyRequests)
if err != nil {
if strings.Contains(err.Error(), "notification_webhooks_profile_name_key") {
return ErrWebhookNameTaken
@@ -161,12 +165,14 @@ func (r *WebhookRepository) Update(ctx context.Context, hook Webhook) error {
signing_secret_ciphertext = $6, enabled = $7,
notify_favorites = $8, notify_watchlist = $9,
notify_continue_watching = $10, notify_next_up = $11,
consecutive_failures = $12, disabled_reason = $13,
notify_requests = $12,
consecutive_failures = $13, disabled_reason = $14,
updated_at = now()
WHERE id = $1`,
hook.ID, hook.Name, hook.Type, hook.URLCiphertext, hook.URLHost,
hook.SigningSecretCiphertext, hook.Enabled,
hook.NotifyFavorites, hook.NotifyWatchlist, hook.NotifyContinueWatching, hook.NotifyNextUp,
hook.NotifyRequests,
hook.ConsecutiveFailures, hook.DisabledReason)
if err != nil {
if strings.Contains(err.Error(), "notification_webhooks_profile_name_key") {
+9 -18
View File
@@ -9,7 +9,6 @@ import (
"sync"
"time"
evt "github.com/Silo-Server/silo-server/internal/events"
"github.com/Silo-Server/silo-server/internal/secret"
"github.com/oklog/ulid/v2"
)
@@ -69,9 +68,10 @@ type webhookSender struct {
cipher *secret.Cipher
settings *Settings
client *http.Client
hub *evt.Hub
payload func(ctx context.Context, row DeliveryRow) DeliveryRowPayload
logger *slog.Logger
// operational posts the auto-disable notice through the system's shared
// durable dispatch path. Wired by NewSystem after construction.
operational func(ctx context.Context, delivery Delivery, opts OperationalDispatch) (*InsertedDelivery, error)
logger *slog.Logger
}
func newWebhookSender(
@@ -79,14 +79,12 @@ func newWebhookSender(
deliveries *DeliveryRepository,
cipher *secret.Cipher,
settings *Settings,
hub *evt.Hub,
) *webhookSender {
sender := &webhookSender{
webhooks: webhooks,
deliveries: deliveries,
cipher: cipher,
settings: settings,
hub: hub,
logger: slog.Default().With("component", "notifications.webhooks"),
}
sender.client = newWebhookHTTPClient(func() bool {
@@ -267,20 +265,13 @@ func (s *webhookSender) disableWebhook(ctx context.Context, hook *Webhook, resul
Type: DeliveryTypeWebhookAutoDisabled,
ReasonFlags: noticeFlags,
}
inserted, err := s.deliveries.InsertOperational(ctx, notice)
if err != nil || inserted == nil {
s.logger.Warn("webhook auto-disable notice insert failed", "webhook_id", hook.ID, "error", err)
if s.operational == nil {
return
}
notice.ID = inserted.ID
notice.CreatedAt = inserted.CreatedAt
if s.hub != nil {
payload := PayloadForRow(DeliveryRow{Delivery: notice})
if s.payload != nil {
payload = s.payload(ctx, DeliveryRow{Delivery: notice})
}
_ = s.hub.PublishJSON(ctx, evt.ChannelNotifications, EventNotificationCreated,
payload, evt.PublishOptions{UserID: hook.UserID, ProfileID: hook.ProfileID})
// Nil WebhookFilter: the notice must never re-dispatch as a webhook, or a
// broken webhook would loop forever.
if _, err := s.operational(ctx, notice, OperationalDispatch{}); err != nil {
s.logger.Warn("webhook auto-disable notice dispatch failed", "webhook_id", hook.ID, "error", err)
}
}
@@ -46,6 +46,7 @@ type WebhookInput struct {
NotifyWatchlist *bool
NotifyContinueWatching *bool
NotifyNextUp *bool
NotifyRequests *bool
}
func validateWebhookName(name string) (string, error) {
@@ -130,6 +131,7 @@ func (s *WebhookService) Create(ctx context.Context, userID int, profileID strin
NotifyWatchlist: boolOrDefault(input.NotifyWatchlist, true),
NotifyContinueWatching: boolOrDefault(input.NotifyContinueWatching, true),
NotifyNextUp: boolOrDefault(input.NotifyNextUp, true),
NotifyRequests: boolOrDefault(input.NotifyRequests, true),
}
hook.URLCiphertext, err = s.cipher.Encrypt(rawURL, webhookURLAAD(hook.ID))
if err != nil {
@@ -215,6 +217,9 @@ func (s *WebhookService) Update(ctx context.Context, profileID, id string, input
if input.NotifyNextUp != nil {
hook.NotifyNextUp = *input.NotifyNextUp
}
if input.NotifyRequests != nil {
hook.NotifyRequests = *input.NotifyRequests
}
if err := s.repo.Update(ctx, *hook); err != nil {
if errors.Is(err, ErrWebhookNameTaken) {
+1
View File
@@ -46,6 +46,7 @@ type Webhook struct {
NotifyWatchlist bool
NotifyContinueWatching bool
NotifyNextUp bool
NotifyRequests bool
ConsecutiveFailures int
DisabledReason *string
LastSuccessAt *time.Time
@@ -32,6 +32,26 @@ func TestBuildWebPushPayload(t *testing.T) {
}
})
t.Run("request fulfilled deep-links to the matched item", func(t *testing.T) {
raw, err := buildWebPushPayload(requestFulfilledTestRow(), "https://cdn.example.com/poster.jpg")
if err != nil {
t.Fatal(err)
}
var payload webPushPayload
if err := json.Unmarshal(raw, &payload); err != nil {
t.Fatal(err)
}
if payload.Title != "Dune is now available" {
t.Fatalf("unexpected title %q", payload.Title)
}
if payload.URL != "/item/movie-123" {
t.Fatalf("unexpected url %q", payload.URL)
}
if payload.Icon != "https://cdn.example.com/poster.jpg" {
t.Fatalf("unexpected icon %q", payload.Icon)
}
})
t.Run("webhook auto-disable routes to settings", func(t *testing.T) {
row := DeliveryRow{Delivery: Delivery{ID: "01X", Type: DeliveryTypeWebhookAutoDisabled}}
raw, err := buildWebPushPayload(row, "")
+11
View File
@@ -81,6 +81,17 @@ func buildWebPushPayload(row DeliveryRow, posterURL string) ([]byte, error) {
payload.URL = "/item/" + *row.EpisodeID
}
payload.Icon = posterURL
case DeliveryTypeRequestFulfilled:
if row.SeriesTitle != "" {
payload.Title = row.SeriesTitle + " is now available"
} else {
payload.Title = "Your request is now available"
}
payload.Body = "Your media request has arrived in the library."
if row.SeriesID != nil {
payload.URL = "/item/" + *row.SeriesID
}
payload.Icon = posterURL
case DeliveryTypeWebhookAutoDisabled:
payload.Title = "A webhook stopped working"
payload.Body = "Open notification settings to fix it."
+66
View File
@@ -0,0 +1,66 @@
package requests
import (
"context"
"log/slog"
)
// FulfillmentNotifier delivers the request.fulfilled notification once a
// completed request's media is confirmed present in the catalog. contentID is
// the matched catalog item id. Implementations must tolerate repeat calls for
// the same request (delivery creation is idempotent); returning nil means the
// request counts as handled and will not be retried.
type FulfillmentNotifier interface {
NotifyFulfilled(ctx context.Context, req Request, contentID string) error
}
// SetFulfillmentNotifier wires the notification system into the reconcile
// service. Optional; without it completed requests are never notified.
func (s *Service) SetFulfillmentNotifier(n FulfillmentNotifier) { s.notifier = n }
// notifyFulfilledLimit bounds one notification pass; the remainder lands on
// the next reconcile run.
const notifyFulfilledLimit = 100
// notifyFulfilledPending notifies completed requests whose media has arrived
// in the catalog. Requests completed by an integration before the library
// scan imports the files stay pending (fulfilled_notified_at IS NULL) and are
// re-checked every run until presence confirms — the notification means
// "watchable in Silo", not "download finished". The delivery insert is
// idempotent (partial unique index per request), so the notify-then-stamp
// ordering can never double-send: a crash between the two retries into a
// dedupe no-op.
func (s *Service) notifyFulfilledPending(ctx context.Context) {
if s.notifier == nil {
return
}
candidates, err := s.store.ListFulfilledUnnotified(ctx, notifyFulfilledLimit)
if err != nil {
slog.WarnContext(ctx, "request fulfill-notify: list candidates failed", "err", err)
return
}
for _, req := range candidates {
if ctx.Err() != nil {
return
}
matches, err := s.lookupPresence(ctx, req.MediaType, []PresenceCandidate{requestPresenceCandidate(*req)})
if err != nil {
slog.WarnContext(ctx, "request fulfill-notify: presence lookup failed",
"request_id", req.ID, "tmdb_id", req.TMDBID, "err", err)
continue
}
match := matches[req.TMDBID]
if !match.Available {
continue // not in the catalog yet; retry next run
}
if err := s.notifier.NotifyFulfilled(ctx, *req, match.ContentID); err != nil {
slog.WarnContext(ctx, "request fulfill-notify: dispatch failed",
"request_id", req.ID, "err", err)
continue
}
if err := s.store.MarkFulfilledNotified(ctx, req.ID); err != nil {
slog.WarnContext(ctx, "request fulfill-notify: mark failed",
"request_id", req.ID, "err", err)
}
}
}
+101
View File
@@ -0,0 +1,101 @@
package requests
import (
"context"
"errors"
"testing"
)
type fakeNotifier struct {
requestIDs []string
contentIDs []string
err error
}
func (f *fakeNotifier) NotifyFulfilled(_ context.Context, req Request, contentID string) error {
if f.err != nil {
return f.err
}
f.requestIDs = append(f.requestIDs, req.ID)
f.contentIDs = append(f.contentIDs, contentID)
return nil
}
func completedRequestFixture(id string, tmdbID int) *Request {
return &Request{
ID: id,
MediaType: MediaTypeMovie,
TMDBID: tmdbID,
Title: "Fixture Movie",
Status: StatusCompleted,
Outcome: OutcomeActive,
RequestedByUserID: 7,
RequestedByProfileID: "profile-1",
}
}
func TestNotifyFulfilledPendingNotifiesAndMarks(t *testing.T) {
store := newFakeStore()
store.requests["req1"] = completedRequestFixture("req1", 42)
store.unnotified = []string{"req1"}
presence := &fakePresence{available: map[MediaType]map[int]bool{
MediaTypeMovie: {42: true},
}}
notifier := &fakeNotifier{}
service := NewService(store, &fakeTMDBClient{}, presence)
service.SetFulfillmentNotifier(notifier)
service.notifyFulfilledPending(context.Background())
if len(notifier.requestIDs) != 1 || notifier.requestIDs[0] != "req1" {
t.Fatalf("expected one notification for req1, got %v", notifier.requestIDs)
}
if want := fakePresenceContentID(MediaTypeMovie, 42); notifier.contentIDs[0] != want {
t.Fatalf("expected content id %q, got %q", want, notifier.contentIDs[0])
}
if len(store.notified) != 1 || store.notified[0] != "req1" {
t.Fatalf("expected req1 marked notified, got %v", store.notified)
}
}
func TestNotifyFulfilledPendingWaitsForCatalogMatch(t *testing.T) {
store := newFakeStore()
store.requests["req1"] = completedRequestFixture("req1", 42)
store.unnotified = []string{"req1"}
notifier := &fakeNotifier{}
service := NewService(store, &fakeTMDBClient{}, &fakePresence{})
service.SetFulfillmentNotifier(notifier)
service.notifyFulfilledPending(context.Background())
if len(notifier.requestIDs) != 0 {
t.Fatalf("expected no notification before catalog match, got %v", notifier.requestIDs)
}
if len(store.notified) != 0 {
t.Fatalf("expected request to stay pending, got marked %v", store.notified)
}
if len(store.unnotified) != 1 {
t.Fatalf("expected request to remain in the pending set")
}
}
func TestNotifyFulfilledPendingRetriesAfterNotifierError(t *testing.T) {
store := newFakeStore()
store.requests["req1"] = completedRequestFixture("req1", 42)
store.unnotified = []string{"req1"}
presence := &fakePresence{available: map[MediaType]map[int]bool{
MediaTypeMovie: {42: true},
}}
notifier := &fakeNotifier{err: errors.New("dispatch failed")}
service := NewService(store, &fakeTMDBClient{}, presence)
service.SetFulfillmentNotifier(notifier)
service.notifyFulfilledPending(context.Background())
if len(store.notified) != 0 {
t.Fatalf("a failed dispatch must not mark the request notified, got %v", store.notified)
}
if len(store.unnotified) != 1 {
t.Fatalf("expected request to remain pending for the next run")
}
}
+47
View File
@@ -374,6 +374,53 @@ func (r *Repository) ListReconciliationCandidates(ctx context.Context, limit int
return out, nil
}
// ListFulfilledUnnotified returns completed requests whose fulfillment
// notification has not fired, oldest first. The horizon bounds how long a
// completed request keeps being presence-polled when its media never appears
// in the catalog (requests completed before the feature shipped are stamped
// by the migration backfill, so the horizon is defense in depth).
func (r *Repository) ListFulfilledUnnotified(ctx context.Context, limit int) ([]*Request, error) {
if limit <= 0 || limit > 500 {
limit = 100
}
rows, err := r.pool.Query(ctx, requestSelectSQL()+`
WHERE outcome = 'active'
AND status = 'completed'
AND fulfilled_notified_at IS NULL
AND completed_at > now() - interval '30 days'
ORDER BY completed_at ASC
LIMIT $1
`, limit)
if err != nil {
return nil, fmt.Errorf("list fulfilled unnotified requests: %w", err)
}
defer rows.Close()
var out []*Request
for rows.Next() {
req, err := scanRequest(rows)
if err != nil {
return nil, err
}
out = append(out, req)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate fulfilled unnotified requests: %w", err)
}
return out, nil
}
// MarkFulfilledNotified stamps the fulfillment-notification marker. Idempotent.
func (r *Repository) MarkFulfilledNotified(ctx context.Context, id string) error {
_, err := r.pool.Exec(ctx, `
UPDATE media_requests SET fulfilled_notified_at = now()
WHERE id = $1 AND fulfilled_notified_at IS NULL`, id)
if err != nil {
return fmt.Errorf("mark request fulfill-notified: %w", err)
}
return nil
}
func (r *Repository) ListMine(ctx context.Context, userID int, filter ListFilter) ([]*Request, error) {
sqlText, args := buildRequestListSQL("requested_by_user_id = $1", []any{userID}, filter)
return r.listRequests(ctx, sqlText, args)
+7
View File
@@ -47,6 +47,7 @@ type Service struct {
router RequestRouterProvider
entitlements EntitlementResolver
requesterIdentity RequesterIdentityResolver
notifier FulfillmentNotifier
Now func() time.Time
}
@@ -719,6 +720,12 @@ func (s *Service) ReconcileRequests(ctx context.Context, limit int) (ReconcileRe
result.Skipped++
}
}
// Presence-gated fulfillment notifications: completion above (and via the
// per-target aggregate path) only marks status; the notification fires
// once the media is confirmed present in the catalog.
if s.notifier != nil {
s.notifyFulfilledPending(ctx)
}
return result, nil
}
+29
View File
@@ -1308,6 +1308,8 @@ type fakeStore struct {
requests map[string]*Request
targets map[string][]Target
targetSeq int64
unnotified []string
notified []string
listIntegrationsCalls int
getSettingsCalls int
@@ -1453,6 +1455,33 @@ func (f *fakeStore) ListReconciliationCandidates(context.Context, int) ([]*Reque
return f.candidates, nil
}
func (f *fakeStore) ListFulfilledUnnotified(context.Context, int) ([]*Request, error) {
f.mu.Lock()
defer f.mu.Unlock()
out := make([]*Request, 0, len(f.unnotified))
for _, id := range f.unnotified {
if req := f.requests[id]; req != nil {
copy := *req
out = append(out, &copy)
}
}
return out, nil
}
func (f *fakeStore) MarkFulfilledNotified(_ context.Context, id string) error {
f.mu.Lock()
defer f.mu.Unlock()
kept := f.unnotified[:0]
for _, pending := range f.unnotified {
if pending != id {
kept = append(kept, pending)
}
}
f.unnotified = kept
f.notified = append(f.notified, id)
return nil
}
func (f *fakeStore) ListMine(context.Context, int, ListFilter) ([]*Request, error) {
f.mu.Lock()
defer f.mu.Unlock()
+6
View File
@@ -18,6 +18,12 @@ type Store interface {
CreateRequest(ctx context.Context, input CreateRequestRecord) (*Request, error)
GetRequest(ctx context.Context, id string) (*Request, error)
ListReconciliationCandidates(ctx context.Context, limit int) ([]*Request, error)
// ListFulfilledUnnotified returns completed requests whose fulfillment
// notification has not fired yet (presence-gated notify pass).
ListFulfilledUnnotified(ctx context.Context, limit int) ([]*Request, error)
// MarkFulfilledNotified stamps a request's fulfillment-notification
// marker so the notify pass stops considering it. Idempotent.
MarkFulfilledNotified(ctx context.Context, id string) error
ListMine(ctx context.Context, userID int, filter ListFilter) ([]*Request, error)
ListAdmin(ctx context.Context, filter ListFilter) ([]*Request, error)
SetStatus(ctx context.Context, id string, status Status, actor Viewer) (*Request, error)
@@ -0,0 +1,43 @@
-- +goose Up
-- +goose StatementBegin
-- request.fulfilled notifications (docs/superpowers/plans/notifications/06,
-- item 2): notify the requesting profile once its media request is actually
-- present in the catalog.
-- Notify marker. NULL means "completed but the fulfillment notification has
-- not fired yet"; the reconcile task keeps presence-checking such requests and
-- stamps this after the delivery is created (or suppressed by preferences).
ALTER TABLE public.media_requests
ADD COLUMN fulfilled_notified_at timestamptz;
-- Flood safety: requests completed before this feature shipped must never
-- notify. Stamp them as already handled so only future completions fire.
UPDATE public.media_requests
SET fulfilled_notified_at = COALESCE(completed_at, now())
WHERE status = 'completed';
-- The reconcile task scans for pending notifications on every run.
CREATE INDEX media_requests_fulfill_notify_idx
ON public.media_requests (completed_at)
WHERE status = 'completed' AND fulfilled_notified_at IS NULL;
-- Per-webhook opt-out for request notifications. Defaults on: every channel
-- the requester configured should tell them (06, item 2).
ALTER TABLE public.notification_webhooks
ADD COLUMN notify_requests boolean NOT NULL DEFAULT true;
-- At-most-once per (profile, request): the operational insert path uses
-- ON CONFLICT DO NOTHING, so a reconcile crash-retry or multi-node race
-- dedupes here instead of double-notifying.
CREATE UNIQUE INDEX notification_deliveries_profile_request_key
ON public.notification_deliveries (profile_id, (reason_flags->>'request_id'))
WHERE type = 'request.fulfilled';
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
DROP INDEX IF EXISTS public.notification_deliveries_profile_request_key;
ALTER TABLE public.notification_webhooks DROP COLUMN IF EXISTS notify_requests;
DROP INDEX IF EXISTS public.media_requests_fulfill_notify_idx;
ALTER TABLE public.media_requests DROP COLUMN IF EXISTS fulfilled_notified_at;
-- +goose StatementEnd
+7
View File
@@ -2225,10 +2225,15 @@ export type EventChannel =
| "notifications";
export interface NotificationReasonFlags {
// episode.available reasons
favorite?: boolean;
watchlist?: boolean;
continue_watching?: boolean;
next_up?: boolean;
// request.fulfilled operational payload
request_id?: string;
tmdb_id?: number;
media_type?: string;
}
export interface AppNotification {
@@ -2292,6 +2297,7 @@ export interface NotificationWebhook {
notify_watchlist: boolean;
notify_continue_watching: boolean;
notify_next_up: boolean;
notify_requests: boolean;
consecutive_failures: number;
disabled_reason: string | null;
last_success_at: string | null;
@@ -2311,6 +2317,7 @@ export interface NotificationWebhookInput {
notify_watchlist?: boolean;
notify_continue_watching?: boolean;
notify_next_up?: boolean;
notify_requests?: boolean;
}
export interface NotificationWebhookTestResult {
@@ -529,6 +529,13 @@ export function RealtimeEventsProvider({ children }: { children: ReactNode }) {
toast(`New episode of ${notification.series_title}`, {
description: [episodeCode, notification.episode_title].filter(Boolean).join(" — "),
});
} else if (notification.type === "request.fulfilled") {
toast(
notification.series_title
? `${notification.series_title} is now available`
: "Your request is now available",
{ description: "Your media request has arrived in the library." },
);
}
return;
}
+11
View File
@@ -46,6 +46,9 @@ function notificationTitle(notification: AppNotification): string {
if (notification.type === "episode.available") {
return notification.series_title || "New episode available";
}
if (notification.type === "request.fulfilled") {
return notification.series_title || "Request available";
}
// Unknown types render with a generic fallback by design — the type
// registry is extensible.
return "Notification";
@@ -58,6 +61,14 @@ function notificationDescription(notification: AppNotification): string {
[code, notification.episode_title].filter(Boolean).join(" — ") || "New episode available"
);
}
if (notification.type === "request.fulfilled") {
const mediaType = notification.reason_flags?.media_type;
return mediaType === "movie"
? "Your requested movie is now available"
: mediaType === "series"
? "Your requested series is now available"
: "Your request is now available";
}
return notification.type;
}
@@ -68,7 +68,14 @@ const REASON_FIELDS = [
{ key: "notify_next_up", label: "Next Up" },
] as const;
type ReasonKey = (typeof REASON_FIELDS)[number]["key"];
// Webhooks additionally carry the request.fulfilled toggle; it is not an
// episode reason, so the profile preferences section keeps REASON_FIELDS.
const WEBHOOK_NOTIFY_FIELDS = [
...REASON_FIELDS,
{ key: "notify_requests", label: "Requests" },
] as const;
type WebhookNotifyKey = (typeof WEBHOOK_NOTIFY_FIELDS)[number]["key"];
function formatRelativeTime(value: string | null): string | null {
if (!value) {
@@ -307,11 +314,12 @@ function WebhookFormDialog({
const update = useUpdateNotificationWebhook();
const [name, setName] = useState(webhook?.name ?? "");
const [url, setUrl] = useState("");
const [reasons, setReasons] = useState<Record<ReasonKey, boolean>>({
const [reasons, setReasons] = useState<Record<WebhookNotifyKey, boolean>>({
notify_favorites: webhook?.notify_favorites ?? true,
notify_watchlist: webhook?.notify_watchlist ?? true,
notify_continue_watching: webhook?.notify_continue_watching ?? true,
notify_next_up: webhook?.notify_next_up ?? true,
notify_requests: webhook?.notify_requests ?? true,
});
const pending = create.isPending || update.isPending;
const editing = webhook != null;
@@ -384,11 +392,14 @@ function WebhookFormDialog({
</div>
<div className="space-y-2">
<Label>Send notifications for</Label>
{REASON_FIELDS.map((field) => {
{WEBHOOK_NOTIFY_FIELDS.map((field) => {
// Requests have no per-profile reason toggle; only the master
// switch suppresses them.
const globallyDisabled =
globalPrefs != null &&
(!globalPrefs.enabled ||
!(globalPrefs[field.key as keyof NotificationPreferences] as boolean));
(field.key !== "notify_requests" &&
!(globalPrefs[field.key as keyof NotificationPreferences] as boolean)));
return (
<div key={field.key} className="flex items-center justify-between gap-3">
<div className="text-sm">
@@ -446,7 +457,7 @@ function WebhookCard({
const failing =
webhook.last_failure_at != null &&
(webhook.last_success_at == null || webhook.last_failure_at > webhook.last_success_at);
const enabledReasons = REASON_FIELDS.filter(
const enabledReasons = WEBHOOK_NOTIFY_FIELDS.filter(
(field) => webhook[field.key as keyof NotificationWebhook] as boolean,
).map((field) => field.label);
@@ -468,7 +479,7 @@ function WebhookCard({
</div>
<div className="text-muted-foreground text-xs">
{enabledReasons.length === REASON_FIELDS.length
{enabledReasons.length === WEBHOOK_NOTIFY_FIELDS.length
? "All reasons"
: enabledReasons.length > 0
? enabledReasons.join(" · ")