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>
129 lines
3.5 KiB
Go
129 lines
3.5 KiB
Go
package notifications
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
// TicketTTL is how long a websocket ticket stays valid. Browsers cannot set
|
|
// custom headers on websocket handshakes, so profile identity is carried by a
|
|
// short-lived single-use ticket in the query string. Reverse-proxy access
|
|
// logs commonly capture query strings; a logged ticket that expired seconds
|
|
// after minting is harmless where a logged profile token would not be.
|
|
const TicketTTL = 30 * time.Second
|
|
|
|
// TicketStore mints and consumes single-use websocket handshake tickets
|
|
// bound to a (user, profile).
|
|
type TicketStore interface {
|
|
Mint(ctx context.Context, userID int, profileID string) (ticket string, ttl time.Duration, err error)
|
|
// Consume validates and invalidates a ticket. ok is false for missing,
|
|
// expired, or already-used tickets.
|
|
Consume(ctx context.Context, ticket string) (userID int, profileID string, ok bool)
|
|
}
|
|
|
|
// NewTicketStore returns a Redis-backed store when a Redis client is
|
|
// available (multi-node websocket serving) and an in-memory store otherwise.
|
|
func NewTicketStore(redisClient *redis.Client) TicketStore {
|
|
if redisClient != nil {
|
|
return &redisTicketStore{client: redisClient}
|
|
}
|
|
return &memoryTicketStore{tickets: make(map[string]memoryTicket)}
|
|
}
|
|
|
|
func newTicketValue() (string, error) {
|
|
raw := make([]byte, 24)
|
|
if _, err := rand.Read(raw); err != nil {
|
|
return "", fmt.Errorf("generate ticket: %w", err)
|
|
}
|
|
return base64.RawURLEncoding.EncodeToString(raw), nil
|
|
}
|
|
|
|
type memoryTicket struct {
|
|
userID int
|
|
profileID string
|
|
expiresAt time.Time
|
|
}
|
|
|
|
type memoryTicketStore struct {
|
|
mu sync.Mutex
|
|
tickets map[string]memoryTicket
|
|
}
|
|
|
|
func (s *memoryTicketStore) Mint(_ context.Context, userID int, profileID string) (string, time.Duration, error) {
|
|
ticket, err := newTicketValue()
|
|
if err != nil {
|
|
return "", 0, err
|
|
}
|
|
now := time.Now()
|
|
s.mu.Lock()
|
|
// Opportunistic sweep keeps the map bounded without a janitor goroutine.
|
|
for key, entry := range s.tickets {
|
|
if now.After(entry.expiresAt) {
|
|
delete(s.tickets, key)
|
|
}
|
|
}
|
|
s.tickets[ticket] = memoryTicket{
|
|
userID: userID,
|
|
profileID: profileID,
|
|
expiresAt: now.Add(TicketTTL),
|
|
}
|
|
s.mu.Unlock()
|
|
return ticket, TicketTTL, nil
|
|
}
|
|
|
|
func (s *memoryTicketStore) Consume(_ context.Context, ticket string) (int, string, bool) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
entry, ok := s.tickets[ticket]
|
|
if !ok {
|
|
return 0, "", false
|
|
}
|
|
delete(s.tickets, ticket) // single-use
|
|
if time.Now().After(entry.expiresAt) {
|
|
return 0, "", false
|
|
}
|
|
return entry.userID, entry.profileID, true
|
|
}
|
|
|
|
type redisTicketStore struct {
|
|
client *redis.Client
|
|
}
|
|
|
|
const redisTicketPrefix = "silo:events:ws-ticket:"
|
|
|
|
func (s *redisTicketStore) Mint(ctx context.Context, userID int, profileID string) (string, time.Duration, error) {
|
|
ticket, err := newTicketValue()
|
|
if err != nil {
|
|
return "", 0, err
|
|
}
|
|
value := strconv.Itoa(userID) + "|" + profileID
|
|
if err := s.client.Set(ctx, redisTicketPrefix+ticket, value, TicketTTL).Err(); err != nil {
|
|
return "", 0, fmt.Errorf("store ticket: %w", err)
|
|
}
|
|
return ticket, TicketTTL, nil
|
|
}
|
|
|
|
func (s *redisTicketStore) Consume(ctx context.Context, ticket string) (int, string, bool) {
|
|
value, err := s.client.GetDel(ctx, redisTicketPrefix+ticket).Result()
|
|
if err != nil {
|
|
return 0, "", false
|
|
}
|
|
parts := strings.SplitN(value, "|", 2)
|
|
if len(parts) != 2 {
|
|
return 0, "", false
|
|
}
|
|
userID, err := strconv.Atoi(parts[0])
|
|
if err != nil || userID <= 0 {
|
|
return 0, "", false
|
|
}
|
|
return userID, parts[1], true
|
|
}
|