Implements the notification system foundation and all v1 delivery channels that need no external infrastructure (specs 00/01/04/05 in docs/superpowers/plans/notifications/): Foundation (spec 01): - episode_availability seeding + per-library seed markers: "newly available" means newly released to this server, so back-catalog imports and first scans never flood (verified on dev: 1.13M episodes seeded silently) - release_events -> profile_series_interest fanout worker with settling delay, per-series burst caps, FOR UPDATE SKIP LOCKED multi-node claims, and a guarded last-notified cursor - interest index maintained via a userstore provider decorator so every favorites/watchlist/progress mutation path (REST, jellycompat, imports, playback) feeds it; progress writes only recompute on state transitions - durable per-profile inbox + read state, forward-sync cursor API, websocket channel with short-lived single-use handshake tickets - web UI: sidebar badge, inbox page, toasts, per-profile preferences - startup/daily tasks: availability seeding, interest rebuild, retention Outbound webhooks (spec 04): - Discord embeds (text-only per the v1 privacy contract) and generic JSON signed Stripe-style with per-webhook secrets - HTTPS-only + private-destination guard enforced at registration and at connect time (DNS-rebinding mitigation); URLs/secrets encrypted at rest - durable per-target outbox enqueued in the fanout transaction, lease-based claims, 24h exponential retry, 3x-consecutive-4xx auto-disable with an in-app notice (loop-guarded) Web push (spec 05): - VAPID keypair self-provisioned at startup (single atomic JSON setting, private half encrypted at rest) — no third-party accounts needed - payloads E2E-encrypted (RFC 8291); 404/410 treated as unsubscribe - service worker + subscribe flow in Settings -> Notifications Shared SMTP core (internal/mail): - feature-agnostic mail.Sender over live email.* settings, STARTTLS or implicit TLS, encrypted password, admin Email settings page with synchronous test send; no consumer yet by design (digest is v1.5) APNs/FCM (specs 02/03) are deferred to v2; the capability endpoint reports them unavailable so clients render truthfully. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
162 lines
4.9 KiB
Go
162 lines
4.9 KiB
Go
package notifications
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
webhookRequestTimeout = 10 * time.Second
|
|
webhookMaxRedirects = 3
|
|
webhookUserAgent = "Silo-Webhook/1.0"
|
|
)
|
|
|
|
// webhookSendResult is the structured outcome of one webhook POST.
|
|
type webhookSendResult struct {
|
|
OK bool
|
|
HTTPStatus int // 0 when no HTTP response was received
|
|
RetryAfter time.Duration // from a 429 Retry-After header, when present
|
|
Duration time.Duration
|
|
// Message is a short, non-sensitive diagnostic suitable for
|
|
// failure_message ("404 Not Found", "dns lookup failed", ...). Never
|
|
// includes URLs or payload contents.
|
|
Message string
|
|
}
|
|
|
|
// newWebhookHTTPClient builds the delivery client: 10s total timeout,
|
|
// non-overridable TLS verification, bounded redirects, and a dialer Control
|
|
// hook that re-validates every resolved address at connect time (DNS
|
|
// rebinding mitigation — the guard runs on the address actually being
|
|
// connected to, each redirect hop included).
|
|
func newWebhookHTTPClient(allowPrivate func() bool) *http.Client {
|
|
dialer := &net.Dialer{
|
|
Timeout: 5 * time.Second,
|
|
Control: func(network, address string, _ syscall.RawConn) error {
|
|
if allowPrivate != nil && allowPrivate() {
|
|
return nil
|
|
}
|
|
host, _, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
return fmt.Errorf("webhook dial: %w", err)
|
|
}
|
|
ip := net.ParseIP(host)
|
|
if ip == nil || !webhookIPAllowed(ip) {
|
|
return errPrivateDestination
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
transport := &http.Transport{
|
|
DialContext: dialer.DialContext,
|
|
ForceAttemptHTTP2: true,
|
|
MaxIdleConns: 16,
|
|
IdleConnTimeout: 60 * time.Second,
|
|
TLSHandshakeTimeout: 5 * time.Second,
|
|
ResponseHeaderTimeout: webhookRequestTimeout,
|
|
}
|
|
return &http.Client{
|
|
Timeout: webhookRequestTimeout,
|
|
Transport: transport,
|
|
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
|
if len(via) >= webhookMaxRedirects {
|
|
return errors.New("too many redirects")
|
|
}
|
|
if req.URL.Scheme != "https" {
|
|
return errors.New("redirect to non-https destination")
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
}
|
|
|
|
var errPrivateDestination = errors.New("destination resolves to a private or special-use network")
|
|
|
|
// sendWebhook POSTs body to url with the given extra headers and classifies
|
|
// the outcome. Success is any 2xx. The response body is drained (bounded) and
|
|
// discarded — Silo never consumes webhook responses for state.
|
|
func sendWebhook(ctx context.Context, client *http.Client, url string, body []byte, headers map[string]string) webhookSendResult {
|
|
started := time.Now()
|
|
result := func(ok bool, status int, message string) webhookSendResult {
|
|
return webhookSendResult{
|
|
OK: ok,
|
|
HTTPStatus: status,
|
|
Duration: time.Since(started),
|
|
Message: message,
|
|
}
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(ctx, webhookRequestTimeout)
|
|
defer cancel()
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
|
|
if err != nil {
|
|
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 {
|
|
req.Header.Set(key, value)
|
|
}
|
|
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return result(false, 0, classifyWebhookError(err))
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10))
|
|
|
|
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
|
return result(true, resp.StatusCode, "")
|
|
}
|
|
out := result(false, resp.StatusCode, http.StatusText(resp.StatusCode))
|
|
if out.Message == "" {
|
|
out.Message = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
|
} else {
|
|
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
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// classifyWebhookError maps transport errors to short diagnostic classes.
|
|
// Messages must stay free of URLs and payload contents.
|
|
func classifyWebhookError(err error) string {
|
|
var dnsErr *net.DNSError
|
|
if errors.As(err, &dnsErr) {
|
|
return "dns lookup failed"
|
|
}
|
|
if errors.Is(err, errPrivateDestination) {
|
|
return "destination resolves to a private network"
|
|
}
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
return "request timed out"
|
|
}
|
|
message := err.Error()
|
|
switch {
|
|
case strings.Contains(message, "tls"):
|
|
return "tls handshake failed"
|
|
case strings.Contains(message, "connection refused"):
|
|
return "connection refused"
|
|
case strings.Contains(message, "too many redirects"):
|
|
return "too many redirects"
|
|
case strings.Contains(message, "redirect to non-https"):
|
|
return "redirect to non-https destination"
|
|
case strings.Contains(message, "timeout") || strings.Contains(message, "deadline"):
|
|
return "request timed out"
|
|
default:
|
|
return "connection failed"
|
|
}
|
|
}
|