Add Apple push notifications support (#255)

* Add push notifications support

* fix(notifications): address push notification review findings

- Gate the capability endpoint's apple_push availability on the admin
  delivery toggle, matching web push: Available now means setup will
  actually deliver.
- Reject direct admin writes to push_relay_deployment_id/api_key; the
  relay issues them as a pair during registration and a lone write
  desyncs them (and poisons the next rotation request).
- Purge a device's registrations under other profiles when it
  re-registers, so a profile switch on a shared device stops the old
  profile's pushes (attempts cascade); adds a DB-backed test.
- Extract the shared channelDispatcher core + retry sweep and rebuild
  the webhook/web push/Apple push dispatchers on it instead of keeping
  three copies of the worker-pool/retry loop.
- Deduplicate relay URL validation (admin setting + register flow) and
  the push outbox attempt-building loops behind shared helpers.
- Cap free-text decline reasons in notification display bodies.
- Fix TestHandleApplePushDisplayDB expectations to match the shared
  display copy (test previously failed under SILO_TEST_DATABASE_URL).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(notifications): route push relay URL writes through registration only

Direct writes to notifications.push_relay_url via the admin settings
endpoint bypassed the relay registration flow, letting the stored URL
drift out of sync with the deployment id / API key pair the relay
minted for it. Reject the URL alongside the deployment id and API key
in the settings handler; POST /admin/notifications/push/relay/register
remains the only path that persists all three together.

The admin UI's Relay URL field now edits local draft state and is
applied by the Register/Rotate action instead of the settings save.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Quick
2026-07-01 17:25:16 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent eafa22a3c3
commit cf0db385f3
32 changed files with 3541 additions and 251 deletions
+15
View File
@@ -2141,6 +2141,21 @@ func (h *AdminHandler) HandleUpdateSetting(w http.ResponseWriter, r *http.Reques
"subtitle_ai.transcribe_quota_period must be day, week, or month")
return
}
case notifications.SettingApplePushDeliveryEnabled:
enabled, err := strconv.ParseBool(strings.TrimSpace(req.Value))
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "notifications.apple_push_delivery_enabled must be true or false")
return
}
req.Value = strconv.FormatBool(enabled)
case notifications.SettingPushRelayURL, notifications.SettingPushRelayDeploymentID, notifications.SettingPushRelayAPIKey:
// The registration flow persists the relay URL, deployment id, and API
// key together; a direct write to any of them desyncs the stored URL
// from the credentials the relay minted for it (and feeds an arbitrary
// id into the next rotation request).
writeError(w, http.StatusBadRequest, "bad_request",
key+" is managed by the push relay registration flow; use POST /admin/notifications/push/relay/register")
return
case catalog.SearchSettingProvider:
switch strings.TrimSpace(strings.ToLower(req.Value)) {
case catalog.SearchProviderPostgres, catalog.SearchProviderMeilisearch:
+279
View File
@@ -0,0 +1,279 @@
package handlers
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
"github.com/Silo-Server/silo-server/internal/notifications"
)
type AdminApplePushHandler struct {
system *notifications.System
settings ServerSettingsStore
client httpDoer
}
type httpDoer interface {
Do(*http.Request) (*http.Response, error)
}
func NewAdminApplePushHandler(system *notifications.System, settings ServerSettingsStore) *AdminApplePushHandler {
return &AdminApplePushHandler{
system: system,
settings: settings,
client: &http.Client{Timeout: 10 * time.Second},
}
}
type adminApplePushTestRequest struct {
ProfileID string `json:"profile_id"`
ServerDeviceID string `json:"server_device_id"`
}
type adminApplePushTestResponse struct {
AttemptID string `json:"attempt_id"`
PushDeviceID string `json:"push_device_id"`
ServerDeviceID string `json:"server_device_id"`
Outcome string `json:"outcome"`
RelayRequestID string `json:"relay_request_id,omitempty"`
UpstreamStatus *int `json:"upstream_status,omitempty"`
UpstreamReason string `json:"upstream_reason,omitempty"`
FailureMessage string `json:"failure_message,omitempty"`
}
type adminPushRelayRegisterRequest struct {
RelayURL string `json:"relay_url"`
}
type pushRelayRegisterRequest struct {
DeploymentID string `json:"deployment_id,omitempty"`
}
type pushRelayRegisterResponse struct {
RequestID string `json:"request_id"`
DeploymentID string `json:"deployment_id"`
APIKey string `json:"api_key"`
KeyPrefix string `json:"key_prefix"`
APNsTopics []string `json:"apns_topics"`
}
type pushRelayNestedError struct {
Error struct {
Code string `json:"code"`
Message string `json:"message"`
RequestID string `json:"request_id"`
} `json:"error"`
}
type adminPushRelayRegisterResponse struct {
RelayURL string `json:"relay_url"`
DeploymentID string `json:"deployment_id"`
KeyPrefix string `json:"key_prefix"`
APIKeyConfigured bool `json:"api_key_configured"`
RelayRequestID string `json:"relay_request_id,omitempty"`
APNsTopics []string `json:"apns_topics,omitempty"`
}
// HandleTest handles POST /admin/notifications/push/apple/test.
func (h *AdminApplePushHandler) HandleTest(w http.ResponseWriter, r *http.Request) {
if h == nil || h.system == nil {
writeError(w, http.StatusServiceUnavailable, "unavailable", "Apple push delivery is not available")
return
}
var req adminApplePushTestRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
result, err := h.system.SendApplePushTest(r.Context(), req.ProfileID, req.ServerDeviceID)
if err != nil {
switch {
case errors.Is(err, notifications.ErrPushDeliveryInvalid):
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
case errors.Is(err, notifications.ErrPushDeliveryNotFound):
writeError(w, http.StatusNotFound, "not_found", "Apple push device not found")
case errors.Is(err, notifications.ErrPushDeliveryUnavailable):
writeError(w, http.StatusServiceUnavailable, "unavailable", "Apple push delivery is not available")
default:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to send Apple push test")
}
return
}
writeJSON(w, http.StatusOK, adminApplePushTestResponse{
AttemptID: result.AttemptID,
PushDeviceID: result.PushDeviceID,
ServerDeviceID: result.ServerDeviceID,
Outcome: result.Outcome,
RelayRequestID: result.RelayRequestID,
UpstreamStatus: result.UpstreamStatus,
UpstreamReason: result.UpstreamReason,
FailureMessage: result.FailureMessage,
})
}
// HandleRegisterRelay handles POST /admin/notifications/push/relay/register.
func (h *AdminApplePushHandler) HandleRegisterRelay(w http.ResponseWriter, r *http.Request) {
if h == nil || h.settings == nil {
writeError(w, http.StatusServiceUnavailable, "unavailable", "Settings store is not available")
return
}
var req adminPushRelayRegisterRequest
dec := json.NewDecoder(r.Body)
dec.DisallowUnknownFields()
if err := dec.Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
relayURL, err := normalizePushRelayURL(req.RelayURL)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
existingDeploymentID, err := h.settings.Get(r.Context(), notifications.SettingPushRelayDeploymentID)
if err != nil {
writeError(w, http.StatusInternalServerError, "settings_error", "Failed to load relay deployment id")
return
}
relayResp, err := h.registerWithRelay(r, relayURL, strings.TrimSpace(existingDeploymentID))
if err != nil {
status, code, message := mapRelayRegistrationError(err)
writeError(w, status, code, message)
return
}
if strings.TrimSpace(relayResp.DeploymentID) == "" || strings.TrimSpace(relayResp.APIKey) == "" {
writeError(w, http.StatusBadGateway, "relay_bad_response", "Push relay returned an incomplete registration response")
return
}
if err := h.settings.Set(r.Context(), notifications.SettingPushRelayURL, relayURL); err != nil {
writeError(w, http.StatusInternalServerError, "settings_error", "Failed to save push relay URL")
return
}
if err := h.settings.Set(r.Context(), notifications.SettingPushRelayDeploymentID, relayResp.DeploymentID); err != nil {
writeError(w, http.StatusInternalServerError, "settings_error", "Failed to save push relay deployment id")
return
}
if err := h.settings.Set(r.Context(), notifications.SettingPushRelayAPIKey, relayResp.APIKey); err != nil {
writeError(w, http.StatusInternalServerError, "settings_error", "Failed to save push relay API key")
return
}
if h.system != nil && h.system.Settings != nil {
h.system.Settings.Invalidate(
notifications.SettingPushRelayURL,
notifications.SettingPushRelayDeploymentID,
notifications.SettingPushRelayAPIKey,
)
}
writeJSON(w, http.StatusOK, adminPushRelayRegisterResponse{
RelayURL: relayURL,
DeploymentID: relayResp.DeploymentID,
KeyPrefix: relayResp.KeyPrefix,
APIKeyConfigured: true,
RelayRequestID: relayResp.RequestID,
APNsTopics: relayResp.APNsTopics,
})
}
func (h *AdminApplePushHandler) registerWithRelay(r *http.Request, relayURL, deploymentID string) (*pushRelayRegisterResponse, error) {
body, err := json.Marshal(pushRelayRegisterRequest{
DeploymentID: deploymentID,
})
if err != nil {
return nil, err
}
httpReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, relayURL+"/v1/deployments/register", bytes.NewReader(body))
if err != nil {
return nil, err
}
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("User-Agent", "Silo-Server/PushRelayRegistration")
resp, err := h.client.Do(httpReq)
if err != nil {
return nil, relayRegistrationError{status: http.StatusBadGateway, code: "relay_unreachable", message: "Push relay could not be reached"}
}
defer func() { _ = resp.Body.Close() }()
data, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<10))
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
var parsed pushRelayNestedError
_ = json.Unmarshal(data, &parsed)
code := parsed.Error.Code
if code == "" {
code = fmt.Sprintf("relay_http_%d", resp.StatusCode)
}
message := parsed.Error.Message
if message == "" {
message = http.StatusText(resp.StatusCode)
}
return nil, relayRegistrationError{status: resp.StatusCode, code: code, message: message}
}
var parsed pushRelayRegisterResponse
if err := json.Unmarshal(data, &parsed); err != nil {
return nil, relayRegistrationError{status: http.StatusBadGateway, code: "relay_bad_response", message: "Push relay returned invalid JSON"}
}
return &parsed, nil
}
type relayRegistrationError struct {
status int
code string
message string
}
func (e relayRegistrationError) Error() string {
return e.code
}
func mapRelayRegistrationError(err error) (int, string, string) {
var relayErr relayRegistrationError
if !errors.As(err, &relayErr) {
return http.StatusInternalServerError, "internal_error", "Failed to register push relay"
}
switch relayErr.status {
case http.StatusForbidden:
return http.StatusUnprocessableEntity, "relay_deployment_rejected", "Push relay rejected this deployment"
case http.StatusTooManyRequests:
return http.StatusTooManyRequests, "relay_rate_limited", relayErr.message
case http.StatusServiceUnavailable:
return http.StatusServiceUnavailable, "relay_unavailable", relayErr.message
default:
if relayErr.status >= 500 {
return http.StatusBadGateway, "relay_error", relayErr.message
}
return http.StatusBadGateway, relayErr.code, relayErr.message
}
}
// normalizePushRelayURLValue trims a relay URL and enforces https with a
// host. Empty input stays empty so callers choose their own default.
func normalizePushRelayURLValue(raw string) (string, error) {
value := strings.TrimRight(strings.TrimSpace(raw), "/")
if value == "" {
return "", nil
}
parsed, err := url.Parse(value)
if err != nil || parsed.Scheme != "https" || parsed.Host == "" {
return "", errors.New("relay_url must be an https URL")
}
return value, nil
}
func normalizePushRelayURL(raw string) (string, error) {
value, err := normalizePushRelayURLValue(raw)
if err != nil {
return "", err
}
if value == "" {
return notifications.DefaultPushRelayURL, nil
}
return value, nil
}
@@ -0,0 +1,123 @@
package handlers
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/Silo-Server/silo-server/internal/notifications"
)
func TestAdminApplePushHandlerUnavailableWithoutSystem(t *testing.T) {
h := NewAdminApplePushHandler(nil, nil)
rec := httptest.NewRecorder()
h.HandleTest(rec, httptest.NewRequest(http.MethodPost, "/admin/notifications/push/apple/test", strings.NewReader(`{}`)))
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("HandleTest without system = %d, want 503", rec.Code)
}
}
func TestAdminApplePushHandlerRejectsInvalidJSON(t *testing.T) {
h := NewAdminApplePushHandler(&notifications.System{}, nil)
rec := httptest.NewRecorder()
h.HandleTest(rec, httptest.NewRequest(http.MethodPost, "/admin/notifications/push/apple/test", strings.NewReader(`{`)))
if rec.Code != http.StatusBadRequest {
t.Fatalf("HandleTest invalid JSON = %d, want 400", rec.Code)
}
}
func TestAdminApplePushHandlerRegistersRelayAndStoresKey(t *testing.T) {
settings := &fakeServerSettingsStore{values: map[string]string{
notifications.SettingPushRelayDeploymentID: "01EXISTING",
}}
var relayReq map[string]string
relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/deployments/register" {
t.Fatalf("relay path = %s", r.URL.Path)
}
if err := json.NewDecoder(r.Body).Decode(&relayReq); err != nil {
t.Fatalf("decode relay request: %v", err)
}
writeJSON(w, http.StatusOK, pushRelayRegisterResponse{
RequestID: "relay-request",
DeploymentID: "01RETURNED",
APIKey: "rk_live_raw-key",
KeyPrefix: "rk_live_raw",
APNsTopics: []string{"org.siloserver.silo"},
})
}))
t.Cleanup(relay.Close)
h := NewAdminApplePushHandler(&notifications.System{Settings: notifications.NewSettings(settings)}, settings)
h.client = relay.Client()
rec := httptest.NewRecorder()
h.HandleRegisterRelay(rec, httptest.NewRequest(http.MethodPost, "/admin/notifications/push/relay/register", strings.NewReader(`{
"relay_url":"`+relay.URL+`"
}`)))
if rec.Code != http.StatusBadRequest {
t.Fatalf("http relay URL status = %d, want 400 because HTTPS is required", rec.Code)
}
httpsRelay := httptest.NewTLSServer(relay.Config.Handler)
t.Cleanup(httpsRelay.Close)
h.client = httpsRelay.Client()
rec = httptest.NewRecorder()
h.HandleRegisterRelay(rec, httptest.NewRequest(http.MethodPost, "/admin/notifications/push/relay/register", strings.NewReader(`{
"relay_url":"`+httpsRelay.URL+`"
}`)))
if rec.Code != http.StatusOK {
t.Fatalf("status = %d (%s), want 200", rec.Code, rec.Body.String())
}
if relayReq["deployment_id"] != "01EXISTING" || len(relayReq) != 1 {
t.Fatalf("relay request = %+v", relayReq)
}
if settings.values[notifications.SettingPushRelayURL] != httpsRelay.URL {
t.Fatalf("stored relay URL = %q", settings.values[notifications.SettingPushRelayURL])
}
if settings.values[notifications.SettingPushRelayDeploymentID] != "01RETURNED" {
t.Fatalf("stored deployment id = %q", settings.values[notifications.SettingPushRelayDeploymentID])
}
if settings.values[notifications.SettingPushRelayAPIKey] != "rk_live_raw-key" {
t.Fatalf("stored api key = %q", settings.values[notifications.SettingPushRelayAPIKey])
}
var body map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("decode response: %v", err)
}
if _, ok := body["api_key"]; ok {
t.Fatal("response leaked raw api_key")
}
if body["api_key_configured"] != true || body["deployment_id"] != "01RETURNED" {
t.Fatalf("response = %+v", body)
}
}
func TestAdminApplePushHandlerMapsRelayRateLimit(t *testing.T) {
settings := &fakeServerSettingsStore{values: map[string]string{}}
relay := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusTooManyRequests, map[string]any{
"error": map[string]string{"code": "rate_limited", "message": "too many deployment registrations from this network"},
})
}))
t.Cleanup(relay.Close)
h := NewAdminApplePushHandler(&notifications.System{}, settings)
h.client = relay.Client()
rec := httptest.NewRecorder()
h.HandleRegisterRelay(rec, httptest.NewRequest(http.MethodPost, "/admin/notifications/push/relay/register", strings.NewReader(`{
"relay_url":"`+relay.URL+`"
}`)))
if rec.Code != http.StatusTooManyRequests {
t.Fatalf("status = %d (%s), want 429", rec.Code, rec.Body.String())
}
if settings.values[notifications.SettingPushRelayAPIKey] != "" {
t.Fatal("api key was stored after relay rejected registration")
}
}
+39 -4
View File
@@ -43,6 +43,8 @@ type notificationSyncResponse struct {
UnreadCount int `json:"unread_count"`
}
type notificationApplePushDisplayResponse = notifications.NotificationDisplay
type unreadCountResponse struct {
Count int `json:"count"`
}
@@ -150,6 +152,29 @@ func (h *NotificationsHandler) HandleGet(w http.ResponseWriter, r *http.Request)
writeJSON(w, http.StatusOK, h.system.PayloadForRow(r.Context(), *row))
}
// HandleApplePushDisplay handles GET /notifications/push/apple/display/{delivery_id}.
// It returns only compact display metadata for notification-service-extension
// enrichment, scoped to the active profile.
func (h *NotificationsHandler) HandleApplePushDisplay(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
profileID := apimw.GetProfileID(r.Context())
id := chi.URLParam(r, "delivery_id")
if id == "" {
writeError(w, http.StatusNotFound, "not_found", "Notification not found")
return
}
row, err := h.system.Deliveries.GetByID(r.Context(), profileID, id)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load notification")
return
}
if row == nil {
writeError(w, http.StatusNotFound, "not_found", "Notification not found")
return
}
writeJSON(w, http.StatusOK, notificationApplePushDisplayResponse(notifications.BuildNotificationDisplay(*row)))
}
// HandleUnreadCount handles GET /notifications/unread-count.
func (h *NotificationsHandler) HandleUnreadCount(w http.ResponseWriter, r *http.Request) {
profileID := apimw.GetProfileID(r.Context())
@@ -318,9 +343,7 @@ type capabilityWebhooks struct {
}
// HandleCapability handles GET /notifications/capability. Clients render
// setup UI from this response instead of introspecting admin settings. Push
// channels report unavailable until they ship
// (docs/superpowers/plans/notifications/02-03).
// setup UI from this response instead of introspecting admin settings.
func (h *NotificationsHandler) HandleCapability(w http.ResponseWriter, r *http.Request) {
webhooks := capabilityWebhooks{Available: false, MaxPerProfile: 0, SupportedTypes: []string{}}
if h.system.Webhooks != nil && h.system.Settings.WebhooksEnabled(r.Context()) {
@@ -364,9 +387,21 @@ func (h *NotificationsHandler) HandleCapability(w http.ResponseWriter, r *http.R
DigestHour: h.system.Settings.DiscordDigestHour(r.Context()),
}
}
applePush := capabilityPush{Available: false, Provider: "off", SupportedModes: []string{"in_app_only"}}
// Like web push above, availability requires both the wiring (cipher/store)
// and the admin delivery toggle: Available must mean "setup will actually
// deliver", not "the server could store a token".
if h.system.PushDevices != nil && h.system.PushDevices.Available() &&
h.system.Settings.ApplePushDeliveryEnabled(r.Context()) {
applePush = capabilityPush{
Available: true,
Provider: notifications.PushProviderSiloRelay,
SupportedModes: []string{notifications.PushModePrivatePush, notifications.PushModeInAppOnly},
}
}
writeJSON(w, http.StatusOK, capabilityResponse{
InApp: capabilityInApp{Enabled: h.system.Settings.UIEnabled(r.Context())},
ApplePush: capabilityPush{Available: false, Provider: "off", SupportedModes: []string{"in_app_only"}},
ApplePush: applePush,
AndroidPush: capabilityPush{Available: false, Provider: "off", SupportedModes: []string{"in_app_only"}},
WebPush: webPush,
Webhooks: webhooks,
@@ -0,0 +1,134 @@
package handlers
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/notifications"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestHandleApplePushDisplayDB(t *testing.T) {
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SILO_TEST_DATABASE_URL to run DB-backed notification display handler test")
}
ctx := context.Background()
config, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatalf("parse db config: %v", err)
}
config.MaxConns = 1
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
t.Fatalf("connect db: %v", err)
}
t.Cleanup(pool.Close)
if _, err := pool.Exec(ctx, `
CREATE TEMP TABLE notification_deliveries (
id text PRIMARY KEY,
release_event_id text,
user_id integer NOT NULL,
profile_id text NOT NULL,
library_id integer,
series_id text,
episode_id text,
type text NOT NULL,
reason_flags jsonb NOT NULL,
status text NOT NULL DEFAULT 'delivered',
read_at timestamptz,
delivered_at timestamptz,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TEMP TABLE episodes (
content_id text PRIMARY KEY,
title text,
season_number integer,
episode_number integer,
overview text
);
CREATE TEMP TABLE media_items (
content_id text PRIMARY KEY,
title text,
poster_path text,
poster_thumbhash text,
poster_source_path text,
type text,
year integer,
overview text,
genres text[],
content_rating text,
rating_imdb double precision,
rating_tmdb double precision,
imdb_id text,
tmdb_id text,
tvdb_id text
);
INSERT INTO media_items (content_id, title, type, genres)
VALUES ('series-1', 'Severance', 'series', ARRAY[]::text[]);
INSERT INTO episodes (content_id, title, season_number, episode_number)
VALUES ('episode-1', 'Hello, Ms. Cobel', 2, 1);
INSERT INTO notification_deliveries (
id, release_event_id, user_id, profile_id, library_id, series_id, episode_id,
type, reason_flags, status
) VALUES (
'delivery-1', 'event-1', 42, 'profile-1', 7, 'series-1', 'episode-1',
'episode.available', '{"favorite":true}', 'delivered'
);
`); err != nil {
t.Fatalf("seed temp tables: %v", err)
}
handler := NewNotificationsHandler(&notifications.System{
Deliveries: notifications.NewDeliveryRepository(pool),
}, nil)
router := chi.NewRouter()
router.Get("/notifications/push/apple/display/{delivery_id}", handler.HandleApplePushDisplay)
req := httptest.NewRequest(http.MethodGet, "/notifications/push/apple/display/delivery-1", nil)
req = req.WithContext(apimw.SetProfileID(req.Context(), "profile-1"))
rr := httptest.NewRecorder()
router.ServeHTTP(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
if got := rr.Header().Get("Cache-Control"); got != "no-store" {
t.Fatalf("Cache-Control = %q", got)
}
var response notifications.NotificationDisplay
if err := json.NewDecoder(rr.Body).Decode(&response); err != nil {
t.Fatalf("decode response: %v", err)
}
if response.DeliveryID != "delivery-1" ||
response.Title != "The latest episode of Severance S02E01 just dropped!" ||
response.Body != "Hello, Ms. Cobel" ||
response.ThreadID != "series:series-1" ||
response.Category != "episode_available" ||
response.URL != "/item/episode-1" {
t.Fatalf("response = %+v", response)
}
req = httptest.NewRequest(http.MethodGet, "/notifications/push/apple/display/delivery-1", nil)
req = req.WithContext(apimw.SetProfileID(req.Context(), "other-profile"))
rr = httptest.NewRecorder()
router.ServeHTTP(rr, req)
if rr.Code != http.StatusNotFound {
t.Fatalf("cross-profile status = %d, body = %s", rr.Code, rr.Body.String())
}
if got := rr.Header().Get("Cache-Control"); got != "no-store" {
t.Fatalf("cross-profile Cache-Control = %q", got)
}
if strings.Contains(rr.Body.String(), "Severance") {
t.Fatalf("cross-profile response leaked display metadata: %s", rr.Body.String())
}
}
@@ -0,0 +1,75 @@
package handlers
import (
"encoding/json"
"errors"
"net/http"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/notifications"
)
type applePushRegisterRequest struct {
DeviceID string `json:"device_id"`
APNsToken string `json:"apns_token"`
APNsEnvironment string `json:"apns_environment"`
APNsTopic string `json:"apns_topic"`
PushMode string `json:"push_mode"`
}
type applePushRegisterResponse struct {
ID string `json:"id"`
ServerDeviceID string `json:"server_device_id"`
Enabled bool `json:"enabled"`
PushMode string `json:"push_mode"`
}
func (h *NotificationsHandler) pushDevices() *notifications.PushDeviceService {
if h == nil || h.system == nil {
return nil
}
return h.system.PushDevices
}
// HandleRegisterApplePushDevice handles POST /devices/push/apple.
func (h *NotificationsHandler) HandleRegisterApplePushDevice(w http.ResponseWriter, r *http.Request) {
service := h.pushDevices()
if service == nil || !service.Available() {
writeError(w, http.StatusServiceUnavailable, "unavailable", "Apple push registration is not available")
return
}
var req applePushRegisterRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
device, err := service.RegisterApple(r.Context(), apimw.GetUserID(r.Context()), apimw.GetProfileID(r.Context()), notifications.ApplePushRegistrationInput{
DeviceID: req.DeviceID,
APNsToken: req.APNsToken,
APNsEnvironment: req.APNsEnvironment,
APNsTopic: req.APNsTopic,
PushMode: req.PushMode,
})
if err != nil {
switch {
case errors.Is(err, notifications.ErrPushDeviceInvalid):
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
case errors.Is(err, notifications.ErrPushDeviceUnsupported):
writeError(w, http.StatusUnprocessableEntity, "unsupported_push_device", err.Error())
case errors.Is(err, notifications.ErrPushDeviceUnavailable):
writeError(w, http.StatusServiceUnavailable, "unavailable", "Apple push registration is not available")
default:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to register Apple push device")
}
return
}
writeJSON(w, http.StatusOK, applePushRegisterResponse{
ID: device.ID,
ServerDeviceID: device.ServerDeviceID,
Enabled: device.Enabled,
PushMode: device.PushMode,
})
}
@@ -0,0 +1,152 @@
package handlers
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/auth"
"github.com/Silo-Server/silo-server/internal/notifications"
"github.com/Silo-Server/silo-server/internal/secret"
)
type handlerPushStore struct {
got notifications.ApplePushDeviceRegistration
calls int
device *notifications.PushDevice
err error
}
func (f *handlerPushStore) UpsertApple(ctx context.Context, registration notifications.ApplePushDeviceRegistration, cipher *secret.Cipher) (*notifications.PushDevice, error) {
f.calls++
f.got = registration
if f.err != nil {
return nil, f.err
}
if f.device != nil {
return f.device, nil
}
return &notifications.PushDevice{
ID: "push-row",
ServerDeviceID: "server-device",
Enabled: true,
PushMode: registration.PushMode,
}, nil
}
func handlerPushCipher(t *testing.T) *secret.Cipher {
t.Helper()
cipher, err := secret.New([]byte("01234567890123456789012345678901"))
if err != nil {
t.Fatalf("new cipher: %v", err)
}
return cipher
}
func newApplePushRequest(body string) *http.Request {
req := httptest.NewRequest(http.MethodPost, "/api/v1/devices/push/apple", strings.NewReader(body))
ctx := apimw.SetClaims(req.Context(), &auth.Claims{UserID: 42, Role: "user", TokenType: auth.TokenTypeAccess})
ctx = apimw.SetProfileID(ctx, "profile-1")
return req.WithContext(ctx)
}
func TestHandleRegisterApplePushDevice(t *testing.T) {
store := &handlerPushStore{}
handler := NewNotificationsHandler(&notifications.System{
PushDevices: notifications.NewPushDeviceService(store, handlerPushCipher(t)),
}, nil)
body := `{
"device_id":"local-device",
"apns_token":"` + strings.Repeat("a", 64) + `",
"apns_environment":"production",
"apns_topic":"org.siloserver.silo",
"push_mode":"private_push"
}`
rr := httptest.NewRecorder()
handler.HandleRegisterApplePushDevice(rr, newApplePushRequest(body))
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
var response applePushRegisterResponse
if err := json.NewDecoder(rr.Body).Decode(&response); err != nil {
t.Fatalf("decode response: %v", err)
}
if response.ID != "push-row" || response.ServerDeviceID != "server-device" || !response.Enabled || response.PushMode != notifications.PushModePrivatePush {
t.Fatalf("unexpected response: %+v", response)
}
if store.calls != 1 {
t.Fatalf("store calls = %d, want 1", store.calls)
}
if store.got.UserID != 42 || store.got.ProfileID != "profile-1" || store.got.DeviceID != "local-device" {
t.Fatalf("unexpected stored registration: %+v", store.got)
}
}
func TestHandleRegisterApplePushDeviceErrors(t *testing.T) {
tests := []struct {
name string
system *notifications.System
body string
wantStatus int
}{
{
name: "service unavailable",
system: &notifications.System{},
body: `{}`,
wantStatus: http.StatusServiceUnavailable,
},
{
name: "invalid json",
system: &notifications.System{
PushDevices: notifications.NewPushDeviceService(&handlerPushStore{}, handlerPushCipher(t)),
},
body: `{`,
wantStatus: http.StatusBadRequest,
},
{
name: "invalid field",
system: &notifications.System{
PushDevices: notifications.NewPushDeviceService(&handlerPushStore{}, handlerPushCipher(t)),
},
body: `{
"device_id":"local-device",
"apns_token":"abcd",
"apns_environment":"production",
"apns_topic":"org.siloserver.silo",
"push_mode":"private_push"
}`,
wantStatus: http.StatusBadRequest,
},
{
name: "unsupported topic",
system: &notifications.System{
PushDevices: notifications.NewPushDeviceService(&handlerPushStore{}, handlerPushCipher(t)),
},
body: `{
"device_id":"local-device",
"apns_token":"` + strings.Repeat("a", 64) + `",
"apns_environment":"production",
"apns_topic":"com.example.app",
"push_mode":"private_push"
}`,
wantStatus: http.StatusUnprocessableEntity,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
handler := NewNotificationsHandler(tt.system, nil)
rr := httptest.NewRecorder()
handler.HandleRegisterApplePushDevice(rr, newApplePushRequest(tt.body))
if rr.Code != tt.wantStatus {
t.Fatalf("status = %d, want %d, body = %s", rr.Code, tt.wantStatus, rr.Body.String())
}
})
}
}
+11
View File
@@ -1626,6 +1626,7 @@ func NewRouter(deps Dependencies) chi.Router {
}
notificationsHandler := handlers.NewNotificationsHandler(deps.Notifications, deps.EventsHub)
r.With(apimw.RequireProfile).Post("/events/ws-ticket", notificationsHandler.HandleMintWSTicket)
r.With(apimw.RequireProfile).Post("/devices/push/apple", notificationsHandler.HandleRegisterApplePushDevice)
// Discord DM channel: the linked identity and mode hang off
// the login account, not a profile, so these stay outside
// the RequireProfile subrouter below (static paths coexist
@@ -1644,6 +1645,7 @@ func NewRouter(deps Dependencies) chi.Router {
r.Get("/capability", notificationsHandler.HandleCapability)
r.Get("/preferences", notificationsHandler.HandleGetPreferences)
r.Put("/preferences", notificationsHandler.HandleUpdatePreferences)
r.Get("/push/apple/display/{delivery_id}", notificationsHandler.HandleApplePushDisplay)
r.Get("/email-preferences", notificationsHandler.HandleGetEmailPreferences)
r.Put("/email-preferences", notificationsHandler.HandleUpdateEmailPreferences)
r.Put("/email-preferences/address", notificationsHandler.HandleRequestEmailAddress)
@@ -2303,6 +2305,15 @@ func NewRouter(deps Dependencies) chi.Router {
if discordNotificationsHandler != nil {
r.Post("/notifications/discord/test", discordNotificationsHandler.HandleAdminTest)
}
if deps.Notifications != nil || settingsRepo != nil {
applePushHandler := handlers.NewAdminApplePushHandler(deps.Notifications, settingsRepo)
if deps.Notifications != nil {
r.Post("/notifications/push/apple/test", applePushHandler.HandleTest)
}
if settingsRepo != nil {
r.Post("/notifications/push/relay/register", applePushHandler.HandleRegisterRelay)
}
}
if deps.Notifications != nil && deps.Notifications.ServerChannels != nil {
serverChannelsHandler := handlers.NewAdminServerChannelsHandler(deps.Notifications)
r.Route("/notifications/server-channels", func(r chi.Router) {
@@ -108,6 +108,9 @@ var SensitiveSettingKeys = map[string]bool{
// value by the notifications system; clients receive the public half via
// the capability endpoint, never from the settings store).
"notifications.web_push.vapid_keypair": true,
// Silo push relay bearer credential for APNs/FCM delivery.
"notifications.push_relay_api_key": true,
}
// EncryptedSettingsRepo decorates a raw settings store, transparently
@@ -158,6 +158,7 @@ func TestSensitiveSettingKeys_Audited(t *testing.T) {
"discord.client_secret",
"discord.bot_token",
"notifications.web_push.vapid_keypair",
"notifications.push_relay_api_key",
}
for _, k := range mustHave {
if !SensitiveSettingKeys[k] {
@@ -0,0 +1,129 @@
package notifications
import (
"context"
"log/slog"
"sync"
"time"
)
// channelDispatcher is the shared dispatch core behind the per-channel
// Dispatcher implementations (webhooks, web push, Apple push). dispatch never
// blocks the fanout loop on destination I/O: it hands the delivery ID to a
// bounded worker pool that claims the delivery's pending outbox attempts and
// sends them. A full queue simply drops the hand-off — the durable pending
// rows are picked up by the retry sweep, so delivery is delayed, never lost.
type channelDispatcher[A any] struct {
channel string // log label, e.g. "web push"
queue chan string
logger *slog.Logger
claimPending func(ctx context.Context, deliveryID string) ([]A, error)
process func(ctx context.Context, attempt A)
// Optional integrated retry sweep: when claimDue is set, run also drains
// due retries and recovers stale pending rows, skipping ticks while
// enabled reports the channel off. Webhooks instead run the standalone
// WebhookRetryWorker, whose lifecycle the system manages separately.
enabled func(ctx context.Context) bool
claimDue func(ctx context.Context, limit int) ([]A, error)
claimLimit int
}
// dispatch queues the delivery's attempts for immediate send without blocking.
func (d *channelDispatcher[A]) dispatch(deliveryID string) {
select {
case d.queue <- deliveryID:
default:
d.logger.Warn(d.channel+" dispatch queue full; deferring to retry worker",
"delivery_id", deliveryID)
}
}
// run consumes the dispatch queue with a bounded worker pool (plus the retry
// sweep when configured) until ctx is canceled. One slow destination cannot
// block other deliveries.
func (d *channelDispatcher[A]) run(ctx context.Context) {
var wg sync.WaitGroup
for range webhookDispatchWorkers {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case deliveryID := <-d.queue:
d.processDelivery(ctx, deliveryID)
}
}
}()
}
if d.claimDue != nil {
wg.Add(1)
go func() {
defer wg.Done()
runRetrySweep(ctx, d.channel, d.logger, d.enabled, d.claimDue, d.claimLimit, d.process)
}()
}
wg.Wait()
}
func (d *channelDispatcher[A]) processDelivery(ctx context.Context, deliveryID string) {
attempts, err := d.claimPending(ctx, deliveryID)
if err != nil {
if ctx.Err() == nil {
d.logger.Warn(d.channel+" attempt claim failed", "delivery_id", deliveryID, "error", err)
}
return
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
d.process(ctx, attempt)
}
}
// runRetrySweep polls for due attempts until ctx is canceled, draining every
// due batch each tick. Shared by the integrated dispatcher sweeps and the
// standalone webhook retry worker.
func runRetrySweep[A any](
ctx context.Context,
channel string,
logger *slog.Logger,
enabled func(context.Context) bool,
claim func(context.Context, int) ([]A, error),
limit int,
process func(context.Context, A),
) {
ticker := time.NewTicker(webhookRetryInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
if !enabled(ctx) {
continue
}
for {
attempts, err := claim(ctx, limit)
if err != nil {
if ctx.Err() == nil {
logger.Warn(channel+" retry claim failed", "error", err)
}
break
}
if len(attempts) == 0 {
break
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
process(ctx, attempt)
}
}
}
}
+139
View File
@@ -0,0 +1,139 @@
package notifications
import (
"fmt"
"strings"
)
// NotificationDisplay is the compact, user-facing display metadata clients use
// when they need a native notification title/body without fetching a full inbox
// page.
type NotificationDisplay struct {
DeliveryID string `json:"delivery_id"`
Title string `json:"title"`
Body string `json:"body,omitempty"`
ThreadID string `json:"thread_id,omitempty"`
Category string `json:"category"`
URL string `json:"url"`
}
// BuildNotificationDisplay renders a delivery into native-notification display
// metadata. Keep this in sync with inbox/websocket semantics by only deriving
// from DeliveryRow.
func BuildNotificationDisplay(row DeliveryRow) NotificationDisplay {
display := NotificationDisplay{
DeliveryID: row.ID,
Title: genericNotificationTitle,
Category: "notification",
URL: "/notifications",
}
switch row.Type {
case DeliveryTypeEpisodeAvailable:
display.Category = "episode_available"
display.Title = episodeDisplayTitle(row)
display.Body = episodeDisplayBody(row)
if row.SeriesID != nil && *row.SeriesID != "" {
display.ThreadID = "series:" + *row.SeriesID
}
if row.EpisodeID != nil && *row.EpisodeID != "" {
display.URL = "/item/" + *row.EpisodeID
}
case DeliveryTypeRequestFulfilled:
display.Category = "request_fulfilled"
display.Title = "Your request is now available"
if row.SeriesTitle != "" {
display.Title = row.SeriesTitle + " is now available"
}
display.Body = "Your media request has arrived in the library."
flags := parseRequestFlags(row.ReasonFlags)
if flags.RequestID != "" {
display.ThreadID = "request:" + flags.RequestID
} else if row.SeriesID != nil && *row.SeriesID != "" {
display.ThreadID = "item:" + *row.SeriesID
}
if row.SeriesID != nil && *row.SeriesID != "" {
display.URL = "/item/" + *row.SeriesID
}
case DeliveryTypeRequestApproved:
flags := parseRequestFlags(row.ReasonFlags)
display.Category = "request_approved"
display.Title = "Your request was approved"
if flags.Title != "" {
display.Title = flags.Title + " was approved"
}
display.Body = "Your media request was approved."
display.ThreadID = requestThreadID(flags)
case DeliveryTypeRequestDeclined:
flags := parseRequestFlags(row.ReasonFlags)
display.Category = "request_declined"
display.Title = "Your request was declined"
if flags.Title != "" {
display.Title = flags.Title + " was declined"
}
display.Body = "Your media request was declined."
if flags.Reason != "" {
// Admin-typed free text with no upstream length cap; keep native
// notification bodies bounded.
display.Body = "Reason: " + truncateDisplayText(flags.Reason, displayBodyMaxLen)
}
display.ThreadID = requestThreadID(flags)
case DeliveryTypeWebhookAutoDisabled:
display.Category = "webhook_auto_disabled"
display.Title = "A webhook stopped working"
display.Body = "Open notification settings to fix it."
display.ThreadID = "settings:notifications"
display.URL = "/settings/notifications"
default:
if row.Type != "" {
display.Category = strings.ReplaceAll(row.Type, ".", "_")
}
}
return display
}
func episodeDisplayTitle(row DeliveryRow) string {
code := episodeDisplayCode(row)
switch {
case row.SeriesTitle != "" && code != "":
return fmt.Sprintf("The latest episode of %s %s just dropped!", row.SeriesTitle, code)
case row.SeriesTitle != "":
return "The latest episode of " + row.SeriesTitle + " just dropped!"
case code != "":
return "New episode " + code + " available"
default:
return "New episode available"
}
}
func episodeDisplayBody(row DeliveryRow) string {
if row.EpisodeTitle != "" {
return row.EpisodeTitle
}
return episodeDisplayCode(row)
}
func episodeDisplayCode(row DeliveryRow) string {
if row.SeasonNumber != nil && row.EpisodeNumber != nil {
return fmt.Sprintf("S%02dE%02d", *row.SeasonNumber, *row.EpisodeNumber)
}
return ""
}
// displayBodyMaxLen bounds free-text notification bodies; matches the varchar
// caps used elsewhere in the delivery pipeline.
const displayBodyMaxLen = 240
func truncateDisplayText(s string, max int) string {
runes := []rune(s)
if len(runes) <= max {
return s
}
return strings.TrimSpace(string(runes[:max-1])) + "…"
}
func requestThreadID(flags RequestFlags) string {
if flags.RequestID == "" {
return ""
}
return "request:" + flags.RequestID
}
+35
View File
@@ -40,6 +40,7 @@ type FanoutWorker struct {
webhooks *WebhookRepository
rateLimiter *profileRateLimiter
webPush *WebPushRepository
pushDevices *PushDeviceRepository
}
// SetWebhookOutbox wires durable webhook attempt enqueueing into the fanout
@@ -55,6 +56,12 @@ func (w *FanoutWorker) SetWebPushOutbox(webPush *WebPushRepository) {
w.webPush = webPush
}
// SetPushOutbox wires durable Apple push attempt enqueueing into the fanout
// transaction.
func (w *FanoutWorker) SetPushOutbox(pushDevices *PushDeviceRepository) {
w.pushDevices = pushDevices
}
// NewFanoutWorker creates a FanoutWorker.
func NewFanoutWorker(
pool *pgxpool.Pool,
@@ -299,6 +306,9 @@ func (w *FanoutWorker) fanOutEvent(ctx context.Context, tx pgx.Tx, event Release
if err := w.enqueueWebPushOutbox(ctx, tx, inserted); err != nil {
return nil, 0, err
}
if err := w.enqueuePushOutbox(ctx, tx, inserted); err != nil {
return nil, 0, err
}
notifiedProfiles := make([]string, 0, len(inserted))
for _, row := range inserted {
@@ -344,6 +354,31 @@ type pendingDelivery struct {
flags ReasonFlags
}
// enqueuePushOutbox inserts pending Apple push attempt rows for each newly
// inserted delivery and enabled private-push device.
func (w *FanoutWorker) enqueuePushOutbox(ctx context.Context, tx pgx.Tx, inserted []InsertedDelivery) error {
if w.pushDevices == nil || len(inserted) == 0 || !w.settings.ApplePushDeliveryEnabled(ctx) {
return nil
}
profileSet := make(map[string]struct{}, len(inserted))
profileIDs := make([]string, 0, len(inserted))
for _, row := range inserted {
if _, ok := profileSet[row.ProfileID]; !ok {
profileSet[row.ProfileID] = struct{}{}
profileIDs = append(profileIDs, row.ProfileID)
}
}
devicesByProfile, err := w.pushDevices.ListEnabledAppleByProfiles(ctx, tx, profileIDs)
if err != nil {
return err
}
attempts := make([]PushDeliveryAttempt, 0, len(inserted))
for _, row := range inserted {
attempts = append(attempts, newPushDeliveryAttempts(row.ID, devicesByProfile[row.ProfileID])...)
}
return w.pushDevices.EnqueuePushAttempts(ctx, tx, attempts)
}
// enqueueWebPushOutbox inserts `pending` web push attempt rows for each newly
// inserted delivery, inside the fanout transaction. Subscriptions have no
// per-reason filters: profile preferences already gated delivery creation,
+12 -2
View File
@@ -18,8 +18,8 @@ type OperationalDispatch struct {
}
// 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
// and the per-target webhook / web push / Apple 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).
@@ -79,6 +79,16 @@ func (s *System) DispatchOperational(ctx context.Context, delivery Delivery, opt
return nil, err
}
}
if s.pushDeviceRepo != nil && s.Settings.ApplePushDeliveryEnabled(ctx) {
devicesByProfile, err := s.pushDeviceRepo.ListEnabledAppleByProfiles(ctx, tx, []string{delivery.ProfileID})
if err != nil {
return nil, err
}
attempts := newPushDeliveryAttempts(row.ID, devicesByProfile[delivery.ProfileID])
if err := s.pushDeviceRepo.EnqueuePushAttempts(ctx, tx, attempts); err != nil {
return nil, err
}
}
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("commit operational dispatch: %w", err)
}
@@ -0,0 +1,157 @@
package notifications
import (
"context"
"io"
"log/slog"
"os"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestDispatchOperationalEnqueuesApplePushAttempts(t *testing.T) {
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SILO_TEST_DATABASE_URL to run DB-backed operational push dispatch test")
}
ctx := context.Background()
config, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatalf("parse db config: %v", err)
}
config.MaxConns = 1
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
t.Fatalf("connect db: %v", err)
}
t.Cleanup(pool.Close)
if _, err := pool.Exec(ctx, `
CREATE TEMP TABLE notification_deliveries (
id text PRIMARY KEY,
release_event_id text,
user_id integer NOT NULL,
profile_id text NOT NULL,
library_id integer,
series_id text,
episode_id text,
type text NOT NULL,
reason_flags jsonb NOT NULL DEFAULT '{}'::jsonb,
status text NOT NULL DEFAULT 'delivered',
read_at timestamptz,
delivered_at timestamptz,
created_at timestamptz NOT NULL DEFAULT now()
) ON COMMIT PRESERVE ROWS;
CREATE TEMP TABLE push_devices (
id text PRIMARY KEY,
user_id integer NOT NULL,
profile_id text NOT NULL,
device_id varchar(128) NOT NULL,
platform text NOT NULL,
provider text NOT NULL,
apns_environment text,
apns_topic text,
apns_token_ciphertext text,
apns_token_hash text,
server_device_id text NOT NULL,
push_mode text NOT NULL DEFAULT 'private_push',
enabled boolean NOT NULL DEFAULT true,
last_seen_at timestamptz,
last_success_at timestamptz,
last_failure_at timestamptz,
last_failure_code text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
) ON COMMIT PRESERVE ROWS;
CREATE TEMP TABLE push_delivery_attempts (
id text PRIMARY KEY,
notification_delivery_id text,
push_device_id text NOT NULL,
trigger_type text NOT NULL,
provider text NOT NULL,
platform text NOT NULL,
attempt_number integer NOT NULL DEFAULT 0,
attempted_at timestamptz,
next_retry_at timestamptz,
outcome text NOT NULL DEFAULT 'pending',
relay_request_id text,
upstream_status integer,
upstream_reason text,
failure_message text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
UNIQUE (notification_delivery_id, push_device_id, trigger_type)
) ON COMMIT PRESERVE ROWS;
`); err != nil {
t.Fatalf("create temp notification push tables: %v", err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO push_devices
(id, user_id, profile_id, device_id, platform, provider, apns_environment, apns_topic,
apns_token_ciphertext, apns_token_hash, server_device_id, push_mode, enabled)
VALUES
('device-private', 42, 'profile-1', 'local-private', 'apple', 'silo_relay', 'sandbox',
'org.siloserver.silo', 'ciphertext', 'hash-1', 'server-private', 'private_push', true),
('device-in-app', 42, 'profile-1', 'local-in-app', 'apple', 'silo_relay', 'sandbox',
'org.siloserver.silo', 'ciphertext', 'hash-2', 'server-in-app', 'in_app_only', true),
('device-disabled', 42, 'profile-1', 'local-disabled', 'apple', 'silo_relay', 'sandbox',
'org.siloserver.silo', 'ciphertext', 'hash-3', 'server-disabled', 'private_push', false),
('device-other-profile', 42, 'profile-2', 'local-other', 'apple', 'silo_relay', 'sandbox',
'org.siloserver.silo', 'ciphertext', 'hash-4', 'server-other', 'private_push', true)
`); err != nil {
t.Fatalf("seed push devices: %v", err)
}
system := &System{
pool: pool,
Settings: NewSettings(mapSettingReader{SettingApplePushDeliveryEnabled: "true"}),
Deliveries: NewDeliveryRepository(pool),
pushDeviceRepo: NewPushDeviceRepository(pool),
dispatcher: NewMultiDispatcher(),
logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
}
inserted, err := system.DispatchOperational(ctx, Delivery{
ID: "delivery-request-1",
UserID: 42,
ProfileID: "profile-1",
Type: DeliveryTypeRequestFulfilled,
ReasonFlags: []byte(`{}`),
}, OperationalDispatch{})
if err != nil {
t.Fatalf("dispatch operational: %v", err)
}
if inserted == nil || inserted.ID != "delivery-request-1" {
t.Fatalf("inserted = %+v", inserted)
}
var deliveryID, pushDeviceID, triggerType, provider, platform, outcome string
if err := pool.QueryRow(ctx, `
SELECT notification_delivery_id, push_device_id, trigger_type, provider, platform, outcome
FROM push_delivery_attempts
`).Scan(&deliveryID, &pushDeviceID, &triggerType, &provider, &platform, &outcome); err != nil {
t.Fatalf("query push attempt: %v", err)
}
if deliveryID != inserted.ID ||
pushDeviceID != "device-private" ||
triggerType != PushTriggerDelivery ||
provider != PushProviderSiloRelay ||
platform != PushPlatformApple ||
outcome != PushOutcomePending {
t.Fatalf("unexpected push attempt: delivery=%q device=%q trigger=%q provider=%q platform=%q outcome=%q",
deliveryID, pushDeviceID, triggerType, provider, platform, outcome)
}
var count int
if err := pool.QueryRow(ctx, `SELECT count(*) FROM push_delivery_attempts`).Scan(&count); err != nil {
t.Fatalf("count push attempts: %v", err)
}
if count != 1 {
t.Fatalf("push attempt count = %d, want 1", count)
}
}
+365
View File
@@ -0,0 +1,365 @@
package notifications
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/jackc/pgx/v5"
"github.com/oklog/ulid/v2"
)
const (
PushTriggerDelivery = "delivery"
PushTriggerTest = "test"
PushOutcomePending = "pending"
PushOutcomeDelivered = "delivered"
PushOutcomeRetrying = "retrying"
PushOutcomeFailed = "failed"
)
var (
ErrPushDeliveryUnavailable = errors.New("apple push delivery unavailable")
ErrPushDeliveryInvalid = errors.New("invalid apple push delivery request")
ErrPushDeliveryNotFound = errors.New("apple push device not found")
)
// PushDeliveryAttempt is one row in the APNs relay outbox/retry log.
type PushDeliveryAttempt struct {
ID string
NotificationDeliveryID *string
PushDeviceID string
TriggerType string
Provider string
Platform string
AttemptNumber int
AttemptedAt time.Time
NextRetryAt *time.Time
Outcome string
RelayRequestID *string
UpstreamStatus *int
UpstreamReason *string
FailureMessage *string
CreatedAt time.Time
UpdatedAt time.Time
}
// newPushDeliveryAttempts builds the pending outbox rows fanning one delivery
// out to its profile's eligible devices. Shared by the fanout worker and
// operational dispatch so both outbox paths enqueue identical rows.
func newPushDeliveryAttempts(deliveryID string, devices []PushDevice) []PushDeliveryAttempt {
attempts := make([]PushDeliveryAttempt, 0, len(devices))
for _, device := range devices {
attempts = append(attempts, PushDeliveryAttempt{
ID: ulid.Make().String(),
NotificationDeliveryID: &deliveryID,
PushDeviceID: device.ID,
TriggerType: PushTriggerDelivery,
})
}
return attempts
}
const pushAttemptReturning = `
RETURNING id, notification_delivery_id, push_device_id, trigger_type, provider, platform,
attempt_number, attempted_at, next_retry_at, outcome, relay_request_id,
upstream_status, upstream_reason, failure_message, created_at, updated_at`
func scanPushDeliveryAttempts(rows pgx.Rows) ([]PushDeliveryAttempt, error) {
defer rows.Close()
attempts := make([]PushDeliveryAttempt, 0, 8)
for rows.Next() {
var attempt PushDeliveryAttempt
if err := rows.Scan(
&attempt.ID,
&attempt.NotificationDeliveryID,
&attempt.PushDeviceID,
&attempt.TriggerType,
&attempt.Provider,
&attempt.Platform,
&attempt.AttemptNumber,
&attempt.AttemptedAt,
&attempt.NextRetryAt,
&attempt.Outcome,
&attempt.RelayRequestID,
&attempt.UpstreamStatus,
&attempt.UpstreamReason,
&attempt.FailureMessage,
&attempt.CreatedAt,
&attempt.UpdatedAt,
); err != nil {
return nil, fmt.Errorf("scan push attempt: %w", err)
}
attempts = append(attempts, attempt)
}
return attempts, rows.Err()
}
// ListEnabledAppleByProfiles loads delivery-eligible APNs devices keyed by profile.
func (r *PushDeviceRepository) ListEnabledAppleByProfiles(ctx context.Context, tx pgx.Tx, profileIDs []string) (map[string][]PushDevice, error) {
out := make(map[string][]PushDevice, len(profileIDs))
if len(profileIDs) == 0 {
return out, nil
}
rows, err := tx.Query(ctx, `SELECT `+pushDeviceColumns+`
FROM push_devices
WHERE profile_id = ANY($1)
AND platform = $2
AND provider = $3
AND push_mode = $4
AND enabled`,
profileIDs, PushPlatformApple, PushProviderSiloRelay, PushModePrivatePush)
if err != nil {
return nil, fmt.Errorf("list enabled push devices: %w", err)
}
defer rows.Close()
for rows.Next() {
device, err := scanPushDevice(rows)
if err != nil {
return nil, fmt.Errorf("scan enabled push device: %w", err)
}
out[device.ProfileID] = append(out[device.ProfileID], *device)
}
return out, rows.Err()
}
// EnqueuePushAttempts inserts pending APNs relay attempts in the fanout transaction.
func (r *PushDeviceRepository) EnqueuePushAttempts(ctx context.Context, tx pgx.Tx, attempts []PushDeliveryAttempt) error {
if len(attempts) == 0 {
return nil
}
var sb strings.Builder
sb.WriteString(`
INSERT INTO push_delivery_attempts
(id, notification_delivery_id, push_device_id, trigger_type, provider, platform, attempt_number, outcome)
VALUES `)
args := make([]any, 0, len(attempts)*8)
for i, attempt := range attempts {
if i > 0 {
sb.WriteString(", ")
}
base := len(args)
sb.WriteString(fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d)",
base+1, base+2, base+3, base+4, base+5, base+6, base+7, base+8))
args = append(args,
attempt.ID,
attempt.NotificationDeliveryID,
attempt.PushDeviceID,
defaultString(attempt.TriggerType, PushTriggerDelivery),
PushProviderSiloRelay,
PushPlatformApple,
0,
PushOutcomePending,
)
}
sb.WriteString(" ON CONFLICT DO NOTHING")
if _, err := tx.Exec(ctx, sb.String(), args...); err != nil {
return fmt.Errorf("enqueue push attempts: %w", err)
}
return nil
}
// EnqueueAppleTestAttempt creates a pending diagnostic attempt for one enabled device.
func (r *PushDeviceRepository) EnqueueAppleTestAttempt(ctx context.Context, profileID, serverDeviceID string) (*PushDeliveryAttempt, *PushDevice, error) {
if r == nil || r.pool == nil {
return nil, nil, ErrPushDeliveryUnavailable
}
profileID = strings.TrimSpace(profileID)
serverDeviceID = strings.TrimSpace(serverDeviceID)
if profileID == "" {
return nil, nil, fmt.Errorf("%w: profile_id is required", ErrPushDeliveryInvalid)
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return nil, nil, fmt.Errorf("begin push test attempt: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
query := `SELECT ` + pushDeviceColumns + `
FROM push_devices
WHERE profile_id = $1
AND platform = $2
AND provider = $3
AND push_mode = $4
AND enabled`
args := []any{profileID, PushPlatformApple, PushProviderSiloRelay, PushModePrivatePush}
if serverDeviceID != "" {
args = append(args, serverDeviceID)
query += fmt.Sprintf(" AND server_device_id = $%d", len(args))
}
query += ` ORDER BY last_seen_at DESC NULLS LAST, created_at DESC LIMIT 1 FOR UPDATE`
device, err := scanPushDevice(tx.QueryRow(ctx, query, args...))
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil, ErrPushDeliveryNotFound
}
if err != nil {
return nil, nil, fmt.Errorf("select push test device: %w", err)
}
attemptID := ulid.Make().String()
rows, err := tx.Query(ctx, `
INSERT INTO push_delivery_attempts
(id, notification_delivery_id, push_device_id, trigger_type, provider, platform, attempt_number, outcome)
VALUES ($1, NULL, $2, $3, $4, $5, 0, $6)`+pushAttemptReturning,
attemptID, device.ID, PushTriggerTest, PushProviderSiloRelay, PushPlatformApple, PushOutcomePending)
if err != nil {
return nil, nil, fmt.Errorf("insert push test attempt: %w", err)
}
attempts, err := scanPushDeliveryAttempts(rows)
if err != nil {
return nil, nil, err
}
if len(attempts) != 1 {
return nil, nil, fmt.Errorf("insert push test attempt returned %d rows", len(attempts))
}
if err := tx.Commit(ctx); err != nil {
return nil, nil, fmt.Errorf("commit push test attempt: %w", err)
}
return &attempts[0], device, nil
}
func (r *PushDeviceRepository) GetPushAttempt(ctx context.Context, id string) (*PushDeliveryAttempt, error) {
rows, err := r.pool.Query(ctx, `SELECT * FROM (
SELECT id, notification_delivery_id, push_device_id, trigger_type, provider, platform,
attempt_number, attempted_at, next_retry_at, outcome, relay_request_id,
upstream_status, upstream_reason, failure_message, created_at, updated_at
FROM push_delivery_attempts
WHERE id = $1
) attempt`, id)
if err != nil {
return nil, fmt.Errorf("get push attempt: %w", err)
}
attempts, err := scanPushDeliveryAttempts(rows)
if err != nil {
return nil, err
}
if len(attempts) == 0 {
return nil, nil
}
return &attempts[0], nil
}
func (r *PushDeviceRepository) getPushDeviceByID(ctx context.Context, id string) (*PushDevice, error) {
device, err := scanPushDevice(r.pool.QueryRow(ctx,
`SELECT `+pushDeviceColumns+` FROM push_devices WHERE id = $1`, id))
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("get push device: %w", err)
}
return device, nil
}
func (r *PushDeviceRepository) ClaimPendingPushForDelivery(ctx context.Context, deliveryID string) ([]PushDeliveryAttempt, error) {
return r.claimPushAttempts(ctx, `
UPDATE push_delivery_attempts SET outcome = 'retrying', next_retry_at = now() + $2, updated_at = now()
WHERE id IN (
SELECT id FROM push_delivery_attempts
WHERE notification_delivery_id = $1 AND outcome = 'pending'
FOR UPDATE SKIP LOCKED
)`+pushAttemptReturning,
deliveryID, webhookClaimLease)
}
func (r *PushDeviceRepository) ClaimPushAttemptByID(ctx context.Context, attemptID string) ([]PushDeliveryAttempt, error) {
return r.claimPushAttempts(ctx, `
UPDATE push_delivery_attempts SET outcome = 'retrying', next_retry_at = now() + $2, updated_at = now()
WHERE id IN (
SELECT id FROM push_delivery_attempts
WHERE id = $1 AND outcome = 'pending'
FOR UPDATE SKIP LOCKED
)`+pushAttemptReturning,
attemptID, webhookClaimLease)
}
func (r *PushDeviceRepository) ClaimDuePushAttempts(ctx context.Context, limit int) ([]PushDeliveryAttempt, error) {
return r.claimPushAttempts(ctx, `
UPDATE push_delivery_attempts SET outcome = 'retrying', next_retry_at = now() + $2, updated_at = now()
WHERE id IN (
SELECT id FROM push_delivery_attempts
WHERE (outcome = 'retrying' AND next_retry_at <= now())
OR (outcome = 'pending' AND attempted_at <= now() - interval '60 seconds')
ORDER BY next_retry_at NULLS FIRST
LIMIT $1
FOR UPDATE SKIP LOCKED
)`+pushAttemptReturning,
limit, webhookClaimLease)
}
func (r *PushDeviceRepository) claimPushAttempts(ctx context.Context, query string, args ...any) ([]PushDeliveryAttempt, error) {
rows, err := r.pool.Query(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("claim push attempts: %w", err)
}
return scanPushDeliveryAttempts(rows)
}
func (r *PushDeviceRepository) FinalizePushAttempt(ctx context.Context, attemptID, outcome string, attemptNumber int, relayRequestID string, upstreamStatus *int, upstreamReason, failureMessage string, nextRetryAt *time.Time) (*PushDeliveryAttempt, error) {
var relayRequestIDPtr *string
if relayRequestID != "" {
relayRequestIDPtr = &relayRequestID
}
var upstreamReasonPtr *string
if upstreamReason != "" {
upstreamReasonPtr = &upstreamReason
}
var failureMessagePtr *string
if failureMessage != "" {
failureMessagePtr = &failureMessage
}
rows, err := r.pool.Query(ctx, `
UPDATE push_delivery_attempts
SET outcome = $2,
attempt_number = $3,
attempted_at = now(),
next_retry_at = $4,
relay_request_id = $5,
upstream_status = $6,
upstream_reason = left($7, 256),
failure_message = left($8, 256),
updated_at = now()
WHERE id = $1`+pushAttemptReturning,
attemptID, outcome, attemptNumber, nextRetryAt, relayRequestIDPtr, upstreamStatus, upstreamReasonPtr, failureMessagePtr)
if err != nil {
return nil, fmt.Errorf("finalize push attempt: %w", err)
}
attempts, err := scanPushDeliveryAttempts(rows)
if err != nil {
return nil, err
}
if len(attempts) == 0 {
return nil, nil
}
return &attempts[0], nil
}
func (r *PushDeviceRepository) RecordPushSuccess(ctx context.Context, deviceID string) error {
_, err := r.pool.Exec(ctx, `
UPDATE push_devices
SET last_success_at = now(), last_failure_at = NULL, last_failure_code = NULL, updated_at = now()
WHERE id = $1`, deviceID)
return err
}
func (r *PushDeviceRepository) RecordPushFailure(ctx context.Context, deviceID, code string, disable bool) error {
_, err := r.pool.Exec(ctx, `
UPDATE push_devices
SET last_failure_at = now(),
last_failure_code = left($2, 128),
enabled = CASE WHEN $3 THEN false ELSE enabled END,
updated_at = now()
WHERE id = $1`, deviceID, code, disable)
return err
}
func defaultString(value, fallback string) string {
if value == "" {
return fallback
}
return value
}
+393
View File
@@ -0,0 +1,393 @@
package notifications
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"regexp"
"strings"
"time"
"github.com/Silo-Server/silo-server/internal/secret"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/oklog/ulid/v2"
)
const (
PushPlatformApple = "apple"
PushProviderSiloRelay = "silo_relay"
PushModeOff = "off"
PushModeInAppOnly = "in_app_only"
PushModePrivatePush = "private_push"
APNsEnvironmentProd = "production"
APNsEnvironmentSandbox = "sandbox"
ApplePushTopicSilo = "org.siloserver.silo"
)
var (
ErrPushDeviceUnavailable = errors.New("push device registration unavailable")
ErrPushDeviceInvalid = errors.New("invalid push device registration")
ErrPushDeviceUnsupported = errors.New("unsupported push device registration")
apnsTokenHexPattern = regexp.MustCompile(`^[0-9a-f]+$`)
)
// PushDevice represents one profile-scoped notification endpoint.
type PushDevice struct {
ID string
UserID int
ProfileID string
DeviceID string
Platform string
Provider string
APNsEnvironment string
APNsTopic string
APNsTokenCiphertext string
APNsTokenHash string
ServerDeviceID string
PushMode string
Enabled bool
LastSeenAt *time.Time
LastSuccessAt *time.Time
LastFailureAt *time.Time
LastFailureCode *string
CreatedAt time.Time
UpdatedAt time.Time
}
type ApplePushRegistrationInput struct {
DeviceID string
APNsToken string
APNsEnvironment string
APNsTopic string
PushMode string
}
type ApplePushDeviceRegistration struct {
UserID int
ProfileID string
DeviceID string
APNsToken string
APNsEnvironment string
APNsTopic string
PushMode string
}
type PushDeviceStore interface {
UpsertApple(ctx context.Context, registration ApplePushDeviceRegistration, cipher *secret.Cipher) (*PushDevice, error)
}
type PushDeviceRepository struct {
pool *pgxpool.Pool
}
func NewPushDeviceRepository(pool *pgxpool.Pool) *PushDeviceRepository {
return &PushDeviceRepository{pool: pool}
}
type PushDeviceService struct {
store PushDeviceStore
cipher *secret.Cipher
}
func NewPushDeviceService(store PushDeviceStore, cipher *secret.Cipher) *PushDeviceService {
return &PushDeviceService{store: store, cipher: cipher}
}
func (s *PushDeviceService) Available() bool {
return s != nil && s.store != nil && s.cipher != nil
}
func (s *PushDeviceService) RegisterApple(ctx context.Context, userID int, profileID string, input ApplePushRegistrationInput) (*PushDevice, error) {
if !s.Available() {
return nil, ErrPushDeviceUnavailable
}
if userID <= 0 {
return nil, fmt.Errorf("%w: user_id is required", ErrPushDeviceInvalid)
}
profileID = strings.TrimSpace(profileID)
if profileID == "" {
return nil, fmt.Errorf("%w: profile_id is required", ErrPushDeviceInvalid)
}
registration, err := normalizeApplePushRegistration(input)
if err != nil {
return nil, err
}
registration.UserID = userID
registration.ProfileID = profileID
return s.store.UpsertApple(ctx, registration, s.cipher)
}
func normalizeApplePushRegistration(input ApplePushRegistrationInput) (ApplePushDeviceRegistration, error) {
deviceID := strings.TrimSpace(input.DeviceID)
if deviceID == "" {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: device_id is required", ErrPushDeviceInvalid)
}
if len(deviceID) > 128 {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: device_id is too long", ErrPushDeviceInvalid)
}
token := strings.ToLower(strings.TrimSpace(input.APNsToken))
if token == "" {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: apns_token is required", ErrPushDeviceInvalid)
}
if len(token) < 64 || len(token) > 256 || !apnsTokenHexPattern.MatchString(token) {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: apns_token must be hex encoded", ErrPushDeviceInvalid)
}
environment := strings.ToLower(strings.TrimSpace(input.APNsEnvironment))
if environment != APNsEnvironmentProd && environment != APNsEnvironmentSandbox {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: apns_environment must be production or sandbox", ErrPushDeviceInvalid)
}
topic := strings.TrimSpace(input.APNsTopic)
if topic == "" {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: apns_topic is required", ErrPushDeviceInvalid)
}
if topic != ApplePushTopicSilo {
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: apns_topic is not supported", ErrPushDeviceUnsupported)
}
pushMode := strings.TrimSpace(input.PushMode)
if pushMode == "" {
pushMode = PushModePrivatePush
}
switch pushMode {
case PushModeOff, PushModeInAppOnly, PushModePrivatePush:
default:
return ApplePushDeviceRegistration{}, fmt.Errorf("%w: push_mode is not supported", ErrPushDeviceUnsupported)
}
return ApplePushDeviceRegistration{
DeviceID: deviceID,
APNsToken: token,
APNsEnvironment: environment,
APNsTopic: topic,
PushMode: pushMode,
}, nil
}
func apnsTokenHash(token string) string {
sum := sha256.Sum256([]byte(strings.ToLower(strings.TrimSpace(token))))
return hex.EncodeToString(sum[:])
}
func pushDeviceAPNsTokenAAD(id string) string {
return secret.RowAAD("push_devices", "apns_token", id)
}
const pushDeviceColumns = `
id,
user_id,
profile_id,
device_id,
platform,
provider,
apns_environment,
apns_topic,
apns_token_ciphertext,
apns_token_hash,
server_device_id,
push_mode,
enabled,
last_seen_at,
last_success_at,
last_failure_at,
last_failure_code,
created_at,
updated_at`
func (r *PushDeviceRepository) UpsertApple(ctx context.Context, registration ApplePushDeviceRegistration, cipher *secret.Cipher) (*PushDevice, error) {
if r == nil || r.pool == nil || cipher == nil {
return nil, ErrPushDeviceUnavailable
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return nil, fmt.Errorf("begin push device upsert: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
// A device install registers for the profile it is currently signed into.
// Purge the same install's registrations under other profiles (attempts
// cascade with them) so a profile switch on a shared device doesn't leave
// the previous profile's notifications flowing to it.
if _, err := tx.Exec(ctx,
`DELETE FROM push_devices WHERE device_id = $1 AND platform = $2 AND profile_id <> $3`,
registration.DeviceID, PushPlatformApple, registration.ProfileID); err != nil {
return nil, fmt.Errorf("purge reassigned push device: %w", err)
}
device, err := r.selectAppleForUpdate(ctx, tx, registration.ProfileID, registration.DeviceID)
if err != nil {
return nil, err
}
if device == nil {
device, err = r.insertApple(ctx, tx, registration, cipher)
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return nil, err
}
if device != nil {
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("commit push device insert: %w", err)
}
return device, nil
}
device, err = r.selectAppleForUpdate(ctx, tx, registration.ProfileID, registration.DeviceID)
if err != nil {
return nil, err
}
if device == nil {
return nil, fmt.Errorf("push device upsert conflict row missing")
}
}
device, err = r.updateApple(ctx, tx, registration, cipher, device)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("commit push device update: %w", err)
}
return device, nil
}
// DeleteAllForProfile removes push registrations for a deleted profile.
func (r *PushDeviceRepository) DeleteAllForProfile(ctx context.Context, profileID string) error {
if r == nil || r.pool == nil {
return nil
}
_, err := r.pool.Exec(ctx, `DELETE FROM push_devices WHERE profile_id = $1`, profileID)
return err
}
func (r *PushDeviceRepository) selectAppleForUpdate(ctx context.Context, tx pgx.Tx, profileID, deviceID string) (*PushDevice, error) {
row := tx.QueryRow(ctx, `SELECT `+pushDeviceColumns+` FROM push_devices WHERE profile_id = $1 AND device_id = $2 AND platform = $3 FOR UPDATE`, profileID, deviceID, PushPlatformApple)
device, err := scanPushDevice(row)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("select push device: %w", err)
}
return device, nil
}
func (r *PushDeviceRepository) insertApple(ctx context.Context, tx pgx.Tx, registration ApplePushDeviceRegistration, cipher *secret.Cipher) (*PushDevice, error) {
id := ulid.Make().String()
serverDeviceID := ulid.Make().String()
ciphertext, err := cipher.Encrypt(registration.APNsToken, pushDeviceAPNsTokenAAD(id))
if err != nil {
return nil, fmt.Errorf("encrypt apns token: %w", err)
}
row := tx.QueryRow(ctx, `
INSERT INTO push_devices (
id,
user_id,
profile_id,
device_id,
platform,
provider,
apns_environment,
apns_topic,
apns_token_ciphertext,
apns_token_hash,
server_device_id,
push_mode,
enabled,
last_seen_at
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, true, now())
ON CONFLICT (profile_id, device_id, platform) DO NOTHING
RETURNING `+pushDeviceColumns,
id,
registration.UserID,
registration.ProfileID,
registration.DeviceID,
PushPlatformApple,
PushProviderSiloRelay,
registration.APNsEnvironment,
registration.APNsTopic,
ciphertext,
apnsTokenHash(registration.APNsToken),
serverDeviceID,
registration.PushMode,
)
device, err := scanPushDevice(row)
if err != nil {
return nil, err
}
return device, nil
}
func (r *PushDeviceRepository) updateApple(ctx context.Context, tx pgx.Tx, registration ApplePushDeviceRegistration, cipher *secret.Cipher, existing *PushDevice) (*PushDevice, error) {
ciphertext, err := cipher.Encrypt(registration.APNsToken, pushDeviceAPNsTokenAAD(existing.ID))
if err != nil {
return nil, fmt.Errorf("encrypt apns token: %w", err)
}
row := tx.QueryRow(ctx, `
UPDATE push_devices
SET user_id = $1,
provider = $2,
apns_environment = $3,
apns_topic = $4,
apns_token_ciphertext = $5,
apns_token_hash = $6,
push_mode = $7,
enabled = true,
last_seen_at = now(),
last_failure_at = NULL,
last_failure_code = NULL,
updated_at = now()
WHERE id = $8
RETURNING `+pushDeviceColumns,
registration.UserID,
PushProviderSiloRelay,
registration.APNsEnvironment,
registration.APNsTopic,
ciphertext,
apnsTokenHash(registration.APNsToken),
registration.PushMode,
existing.ID,
)
device, err := scanPushDevice(row)
if err != nil {
return nil, fmt.Errorf("update push device: %w", err)
}
return device, nil
}
func scanPushDevice(row pgx.Row) (*PushDevice, error) {
var device PushDevice
if err := row.Scan(
&device.ID,
&device.UserID,
&device.ProfileID,
&device.DeviceID,
&device.Platform,
&device.Provider,
&device.APNsEnvironment,
&device.APNsTopic,
&device.APNsTokenCiphertext,
&device.APNsTokenHash,
&device.ServerDeviceID,
&device.PushMode,
&device.Enabled,
&device.LastSeenAt,
&device.LastSuccessAt,
&device.LastFailureAt,
&device.LastFailureCode,
&device.CreatedAt,
&device.UpdatedAt,
); err != nil {
return nil, err
}
return &device, nil
}
+340
View File
@@ -0,0 +1,340 @@
package notifications
import (
"context"
"errors"
"os"
"strings"
"testing"
"github.com/Silo-Server/silo-server/internal/secret"
"github.com/jackc/pgx/v5/pgxpool"
)
type fakePushDeviceStore struct {
got ApplePushDeviceRegistration
calls int
device *PushDevice
err error
}
func (f *fakePushDeviceStore) UpsertApple(ctx context.Context, registration ApplePushDeviceRegistration, cipher *secret.Cipher) (*PushDevice, error) {
f.calls++
f.got = registration
if f.err != nil {
return nil, f.err
}
if f.device != nil {
return f.device, nil
}
return &PushDevice{
ID: "device-row",
ServerDeviceID: "server-device",
Enabled: true,
PushMode: registration.PushMode,
}, nil
}
func testPushCipher(t *testing.T) *secret.Cipher {
t.Helper()
cipher, err := secret.New([]byte("01234567890123456789012345678901"))
if err != nil {
t.Fatalf("new cipher: %v", err)
}
return cipher
}
func validApplePushInput() ApplePushRegistrationInput {
return ApplePushRegistrationInput{
DeviceID: "local-device",
APNsToken: strings.Repeat("a", 64),
APNsEnvironment: APNsEnvironmentProd,
APNsTopic: ApplePushTopicSilo,
PushMode: PushModePrivatePush,
}
}
func TestPushDeviceServiceRegisterAppleNormalizesAndStores(t *testing.T) {
store := &fakePushDeviceStore{}
service := NewPushDeviceService(store, testPushCipher(t))
input := validApplePushInput()
input.DeviceID = " local-device "
input.APNsToken = strings.ToUpper(input.APNsToken)
input.APNsEnvironment = "Production"
input.PushMode = ""
device, err := service.RegisterApple(context.Background(), 42, " profile-1 ", input)
if err != nil {
t.Fatalf("register apple: %v", err)
}
if device.ServerDeviceID != "server-device" || !device.Enabled || device.PushMode != PushModePrivatePush {
t.Fatalf("unexpected device response: %+v", device)
}
if store.calls != 1 {
t.Fatalf("store calls = %d, want 1", store.calls)
}
if store.got.UserID != 42 || store.got.ProfileID != "profile-1" {
t.Fatalf("unexpected owner scope: %+v", store.got)
}
if store.got.DeviceID != "local-device" {
t.Fatalf("device_id = %q", store.got.DeviceID)
}
if store.got.APNsToken != strings.Repeat("a", 64) {
t.Fatalf("apns token was not canonicalized")
}
if store.got.APNsEnvironment != APNsEnvironmentProd {
t.Fatalf("environment = %q", store.got.APNsEnvironment)
}
if store.got.PushMode != PushModePrivatePush {
t.Fatalf("push mode = %q", store.got.PushMode)
}
}
func TestPushDeviceServiceRegisterAppleRejectsInvalidOrUnsupported(t *testing.T) {
tests := []struct {
name string
mutate func(*ApplePushRegistrationInput)
wantErr error
}{
{
name: "missing device id",
mutate: func(input *ApplePushRegistrationInput) { input.DeviceID = " " },
wantErr: ErrPushDeviceInvalid,
},
{
name: "malformed token",
mutate: func(input *ApplePushRegistrationInput) { input.APNsToken = strings.Repeat("z", 64) },
wantErr: ErrPushDeviceInvalid,
},
{
name: "short token",
mutate: func(input *ApplePushRegistrationInput) { input.APNsToken = "abcd" },
wantErr: ErrPushDeviceInvalid,
},
{
name: "invalid environment",
mutate: func(input *ApplePushRegistrationInput) { input.APNsEnvironment = "development" },
wantErr: ErrPushDeviceInvalid,
},
{
name: "unsupported topic",
mutate: func(input *ApplePushRegistrationInput) { input.APNsTopic = "com.example.app" },
wantErr: ErrPushDeviceUnsupported,
},
{
name: "unsupported push mode",
mutate: func(input *ApplePushRegistrationInput) { input.PushMode = "custom_apns" },
wantErr: ErrPushDeviceUnsupported,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
input := validApplePushInput()
tt.mutate(&input)
store := &fakePushDeviceStore{}
service := NewPushDeviceService(store, testPushCipher(t))
_, err := service.RegisterApple(context.Background(), 42, "profile-1", input)
if !errors.Is(err, tt.wantErr) {
t.Fatalf("error = %v, want %v", err, tt.wantErr)
}
if store.calls != 0 {
t.Fatalf("store was called for rejected input")
}
})
}
}
func TestPushDeviceAPNsTokenEncryptionUsesRowAAD(t *testing.T) {
cipher := testPushCipher(t)
token := strings.Repeat("a", 64)
ciphertext, err := cipher.Encrypt(token, pushDeviceAPNsTokenAAD("row-1"))
if err != nil {
t.Fatalf("encrypt token: %v", err)
}
if ciphertext == token || strings.Contains(ciphertext, token) {
t.Fatalf("ciphertext exposes token: %q", ciphertext)
}
plaintext, err := cipher.Decrypt(ciphertext, pushDeviceAPNsTokenAAD("row-1"))
if err != nil {
t.Fatalf("decrypt token: %v", err)
}
if plaintext != token {
t.Fatalf("plaintext = %q, want token", plaintext)
}
if _, err := cipher.Decrypt(ciphertext, pushDeviceAPNsTokenAAD("row-2")); err == nil {
t.Fatalf("decrypt with wrong row AAD succeeded")
}
if got, want := apnsTokenHash(strings.ToUpper(token)), apnsTokenHash(token); got != want {
t.Fatalf("hash should be token-case agnostic: %q != %q", got, want)
}
}
// newPushDeviceTestRepo connects to SILO_TEST_DATABASE_URL (skipping when
// unset) and shadows push_devices with a session-local temp table, pinning the
// pool to one connection so every query sees it.
func newPushDeviceTestRepo(t *testing.T) (*PushDeviceRepository, *pgxpool.Pool) {
t.Helper()
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set SILO_TEST_DATABASE_URL to run DB-backed push device repository test")
}
ctx := context.Background()
config, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatalf("parse db config: %v", err)
}
config.MaxConns = 1
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
t.Fatalf("connect db: %v", err)
}
t.Cleanup(pool.Close)
if _, err := pool.Exec(ctx, `
CREATE TEMP TABLE push_devices (
id text PRIMARY KEY,
user_id integer NOT NULL,
profile_id text NOT NULL,
device_id varchar(128) NOT NULL,
platform text NOT NULL,
provider text NOT NULL,
apns_environment text,
apns_topic text,
apns_token_ciphertext text,
apns_token_hash text,
server_device_id text NOT NULL,
push_mode text NOT NULL DEFAULT 'private_push',
enabled boolean NOT NULL DEFAULT true,
last_seen_at timestamptz,
last_success_at timestamptz,
last_failure_at timestamptz,
last_failure_code text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
CONSTRAINT push_devices_profile_device_platform_key UNIQUE (profile_id, device_id, platform),
CONSTRAINT push_devices_server_device_id_key UNIQUE (server_device_id)
) ON COMMIT PRESERVE ROWS`); err != nil {
t.Fatalf("create temp push_devices table: %v", err)
}
return NewPushDeviceRepository(pool), pool
}
func TestPushDeviceRepositoryUpsertApplePreservesStableIDs(t *testing.T) {
ctx := context.Background()
repo, pool := newPushDeviceTestRepo(t)
cipher := testPushCipher(t)
registration := ApplePushDeviceRegistration{
UserID: 42,
ProfileID: "profile-1",
DeviceID: "local-device",
APNsToken: strings.Repeat("a", 64),
APNsEnvironment: APNsEnvironmentProd,
APNsTopic: ApplePushTopicSilo,
PushMode: PushModePrivatePush,
}
first, err := repo.UpsertApple(ctx, registration, cipher)
if err != nil {
t.Fatalf("first upsert: %v", err)
}
registration.APNsToken = strings.Repeat("b", 64)
registration.PushMode = PushModeInAppOnly
second, err := repo.UpsertApple(ctx, registration, cipher)
if err != nil {
t.Fatalf("second upsert: %v", err)
}
if second.ID != first.ID {
t.Fatalf("row id changed on token rotation: %q != %q", second.ID, first.ID)
}
if second.ServerDeviceID != first.ServerDeviceID {
t.Fatalf("server device id changed on token rotation: %q != %q", second.ServerDeviceID, first.ServerDeviceID)
}
if second.APNsTokenHash != apnsTokenHash(registration.APNsToken) {
t.Fatalf("token hash = %q, want rotated hash", second.APNsTokenHash)
}
if first.APNsTokenHash == second.APNsTokenHash {
t.Fatalf("token hash did not change after rotation")
}
plaintext, err := cipher.Decrypt(second.APNsTokenCiphertext, pushDeviceAPNsTokenAAD(second.ID))
if err != nil {
t.Fatalf("decrypt rotated token: %v", err)
}
if plaintext != registration.APNsToken {
t.Fatalf("rotated plaintext = %q", plaintext)
}
if !second.Enabled || second.PushMode != PushModeInAppOnly {
t.Fatalf("upsert did not re-enable/update mode: %+v", second)
}
var count int
if err := pool.QueryRow(ctx, `SELECT count(*) FROM push_devices`).Scan(&count); err != nil {
t.Fatalf("count rows: %v", err)
}
if count != 1 {
t.Fatalf("row count = %d, want 1", count)
}
}
func TestPushDeviceRepositoryUpsertApplePurgesOtherProfiles(t *testing.T) {
ctx := context.Background()
repo, pool := newPushDeviceTestRepo(t)
cipher := testPushCipher(t)
registration := ApplePushDeviceRegistration{
UserID: 42,
ProfileID: "profile-parent",
DeviceID: "shared-phone",
APNsToken: strings.Repeat("a", 64),
APNsEnvironment: APNsEnvironmentProd,
APNsTopic: ApplePushTopicSilo,
PushMode: PushModePrivatePush,
}
if _, err := repo.UpsertApple(ctx, registration, cipher); err != nil {
t.Fatalf("register under first profile: %v", err)
}
// A different install on another profile must be untouched by the purge.
other := registration
other.ProfileID = "profile-parent"
other.DeviceID = "other-phone"
if _, err := repo.UpsertApple(ctx, other, cipher); err != nil {
t.Fatalf("register unrelated device: %v", err)
}
// The shared phone switches profiles and re-registers: the old profile's
// row for that install must be gone, not left enabled.
registration.ProfileID = "profile-kid"
device, err := repo.UpsertApple(ctx, registration, cipher)
if err != nil {
t.Fatalf("register under second profile: %v", err)
}
if device.ProfileID != "profile-kid" {
t.Fatalf("profile = %q, want profile-kid", device.ProfileID)
}
rows, err := pool.Query(ctx, `SELECT profile_id, device_id FROM push_devices ORDER BY profile_id`)
if err != nil {
t.Fatalf("list rows: %v", err)
}
defer rows.Close()
got := map[string]string{}
for rows.Next() {
var profileID, deviceID string
if err := rows.Scan(&profileID, &deviceID); err != nil {
t.Fatalf("scan row: %v", err)
}
got[profileID] = deviceID
}
if err := rows.Err(); err != nil {
t.Fatalf("rows: %v", err)
}
want := map[string]string{"profile-kid": "shared-phone", "profile-parent": "other-phone"}
if len(got) != len(want) || got["profile-kid"] != want["profile-kid"] || got["profile-parent"] != want["profile-parent"] {
t.Fatalf("rows after reassignment = %v, want %v", got, want)
}
}
+343
View File
@@ -0,0 +1,343 @@
package notifications
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
"time"
"github.com/Silo-Server/silo-server/internal/secret"
)
var pushRetrySchedule = []time.Duration{
0,
30 * time.Second,
2 * time.Minute,
10 * time.Minute,
30 * time.Minute,
}
const (
pushMaxAttempts = 5
pushDispatchQueue = 512
pushRetryClaimLimit = 100
relayAppleSendPath = "/v1/apple/send"
)
func pushRetryDelay(completedAttempt int) (time.Duration, bool) {
if completedAttempt < 1 || completedAttempt >= pushMaxAttempts {
return 0, false
}
return pushRetrySchedule[completedAttempt] - pushRetrySchedule[completedAttempt-1], true
}
type pushRelayAppleRequest struct {
Token string `json:"token"`
Environment string `json:"environment"`
Topic string `json:"topic"`
Mode string `json:"mode"`
ServerDeviceID string `json:"server_device_id"`
DeliveryID string `json:"delivery_id"`
CollapseID *string `json:"collapse_id,omitempty"`
}
type pushRelayAppleResponse struct {
RequestID string `json:"request_id"`
APNsID string `json:"apns_id"`
Status string `json:"status"`
}
type pushRelayErrorResponse struct {
Error struct {
Code string `json:"code"`
Message string `json:"message"`
RequestID string `json:"request_id"`
} `json:"error"`
}
type pushSendResult struct {
OK bool
HTTPStatus int
RetryAfter time.Duration
RelayRequestID string
UpstreamReason string
Message string
TerminalDevice bool
}
type pushSender struct {
devices *PushDeviceRepository
deliveries *DeliveryRepository
cipher *secret.Cipher
settings *Settings
client *http.Client
logger *slog.Logger
}
func newPushSender(devices *PushDeviceRepository, deliveries *DeliveryRepository, cipher *secret.Cipher, settings *Settings) *pushSender {
return &pushSender{
devices: devices,
deliveries: deliveries,
cipher: cipher,
settings: settings,
client: newWebhookHTTPClient(nil),
logger: slog.Default().With("component", "notifications.apple_push"),
}
}
func (s *pushSender) processAttempt(ctx context.Context, attempt PushDeliveryAttempt) *PushDeliveryAttempt {
device, err := s.devices.getPushDeviceByID(ctx, attempt.PushDeviceID)
if err != nil || device == nil {
if err == nil {
return s.finalize(ctx, attempt, PushOutcomeFailed, "", "push device deleted", nil, "", nil)
}
if ctx.Err() == nil {
s.logger.Warn("push device lookup failed", "attempt_id", attempt.ID, "error", err)
}
return nil
}
if !device.Enabled || device.PushMode != PushModePrivatePush || !s.settings.ApplePushDeliveryEnabled(ctx) {
return s.finalize(ctx, attempt, PushOutcomeFailed, "delivery_disabled", "apple push delivery disabled", nil, "", nil)
}
if attempt.NotificationDeliveryID != nil {
row, err := s.deliveries.GetRowByID(ctx, *attempt.NotificationDeliveryID)
if err != nil {
if ctx.Err() == nil {
s.logger.Warn("push delivery lookup failed",
"attempt_id", attempt.ID,
"delivery_id", *attempt.NotificationDeliveryID,
"error", err)
}
return nil
}
if row == nil {
return s.finalize(ctx, attempt, PushOutcomeFailed, "delivery_missing", "delivery row missing", nil, "", nil)
}
if row.ProfileID != device.ProfileID {
return s.finalize(ctx, attempt, PushOutcomeFailed, "device_reassigned", "push device reassigned", nil, "", nil)
}
}
token, err := s.cipher.Decrypt(device.APNsTokenCiphertext, pushDeviceAPNsTokenAAD(device.ID))
if err != nil {
return s.finalize(ctx, attempt, PushOutcomeFailed, "decrypt_failed", "APNs token decrypt failed", nil, "", nil)
}
result := s.send(ctx, attempt, device, token)
attemptNumber := attempt.AttemptNumber + 1
if result.OK {
updated, _ := s.devices.FinalizePushAttempt(ctx, attempt.ID, PushOutcomeDelivered, attemptNumber,
result.RelayRequestID, &result.HTTPStatus, result.UpstreamReason, "", nil)
_ = s.devices.RecordPushSuccess(ctx, device.ID)
return updated
}
statusPtr := (*int)(nil)
if result.HTTPStatus > 0 {
statusPtr = &result.HTTPStatus
}
code := result.UpstreamReason
if code == "" {
code = result.Message
}
if code == "" {
code = "push_delivery_failed"
}
_ = s.devices.RecordPushFailure(ctx, device.ID, code, result.TerminalDevice)
delay, more := pushRetryDelay(attemptNumber)
if result.RetryAfter > 0 {
delay = result.RetryAfter
}
if more && !result.TerminalDevice && retryableHTTPStatus(result.HTTPStatus) {
nextRetry := time.Now().Add(delay)
return s.finalize(ctx, attempt, PushOutcomeRetrying, result.UpstreamReason, result.Message, statusPtr, result.RelayRequestID, &nextRetry)
}
return s.finalize(ctx, attempt, PushOutcomeFailed, result.UpstreamReason, result.Message, statusPtr, result.RelayRequestID, nil)
}
func (s *pushSender) finalize(ctx context.Context, attempt PushDeliveryAttempt, outcome string, reason, message string, statusPtr *int, relayRequestID string, nextRetryAt *time.Time) *PushDeliveryAttempt {
attemptNumber := attempt.AttemptNumber + 1
updated, err := s.devices.FinalizePushAttempt(ctx, attempt.ID, outcome, attemptNumber, relayRequestID, statusPtr, reason, message, nextRetryAt)
if err != nil && ctx.Err() == nil {
s.logger.Warn("finalize push attempt failed", "attempt_id", attempt.ID, "error", err)
}
return updated
}
func (s *pushSender) send(ctx context.Context, attempt PushDeliveryAttempt, device *PushDevice, token string) pushSendResult {
apiKey := s.settings.PushRelayAPIKey(ctx)
if apiKey == "" {
return pushSendResult{Message: "push relay API key not configured", UpstreamReason: "relay_api_key_missing"}
}
relayURL := s.settings.PushRelayURL(ctx)
deliveryID := attempt.ID
if attempt.NotificationDeliveryID != nil {
deliveryID = *attempt.NotificationDeliveryID
}
collapseID := deliveryID
body, err := json.Marshal(pushRelayAppleRequest{
Token: token,
Environment: device.APNsEnvironment,
Topic: device.APNsTopic,
Mode: "private_alert",
ServerDeviceID: device.ServerDeviceID,
DeliveryID: deliveryID,
CollapseID: &collapseID,
})
if err != nil {
return pushSendResult{Message: "relay payload build failed", UpstreamReason: "payload_build_failed"}
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, relayURL+relayAppleSendPath, bytes.NewReader(body))
if err != nil {
return pushSendResult{Message: "invalid push relay URL", UpstreamReason: "invalid_relay_url"}
}
if req.URL.Scheme != schemeHTTPS {
return pushSendResult{Message: "invalid push relay URL", UpstreamReason: "invalid_relay_url"}
}
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "Silo-Push/1.0")
req.Header.Set("Idempotency-Key", fmt.Sprintf("%s:%d", attempt.ID, attempt.AttemptNumber+1))
resp, err := s.client.Do(req)
if err != nil {
return pushSendResult{Message: classifyWebhookError(err), UpstreamReason: "relay_unreachable"}
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
var parsed pushRelayAppleResponse
_ = json.NewDecoder(io.LimitReader(resp.Body, 16<<10)).Decode(&parsed)
if parsed.RequestID == "" {
parsed.RequestID = resp.Header.Get("X-Request-ID")
}
return pushSendResult{
OK: true,
HTTPStatus: resp.StatusCode,
RelayRequestID: parsed.RequestID,
}
}
var parsed pushRelayErrorResponse
data, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<10))
_ = json.Unmarshal(data, &parsed)
code := parsed.Error.Code
if code == "" {
code = fmt.Sprintf("http_%d", resp.StatusCode)
}
message := parsed.Error.Message
if message == "" {
message = http.StatusText(resp.StatusCode)
}
if message == "" {
message = fmt.Sprintf("HTTP %d", resp.StatusCode)
}
return pushSendResult{
HTTPStatus: resp.StatusCode,
RetryAfter: parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()),
RelayRequestID: parsed.Error.RequestID,
UpstreamReason: code,
Message: strings.TrimSpace(message),
TerminalDevice: resp.StatusCode == http.StatusUnprocessableEntity && code == "apns_rejected",
}
}
// PushDispatcher implements the channel Dispatcher interface for Apple push
// on top of the shared channelDispatcher core, with the retry/recovery sweep
// integrated.
type PushDispatcher struct {
core channelDispatcher[PushDeliveryAttempt]
}
func newPushDispatcher(sender *pushSender) *PushDispatcher {
return &PushDispatcher{core: channelDispatcher[PushDeliveryAttempt]{
channel: "apple push",
queue: make(chan string, pushDispatchQueue),
logger: slog.Default().With("component", "notifications.apple_push.dispatch"),
claimPending: sender.devices.ClaimPendingPushForDelivery,
process: func(ctx context.Context, attempt PushDeliveryAttempt) {
sender.processAttempt(ctx, attempt)
},
enabled: sender.settings.ApplePushDeliveryEnabled,
claimDue: sender.devices.ClaimDuePushAttempts,
claimLimit: pushRetryClaimLimit,
}}
}
// Dispatch queues the delivery's Apple push attempts for immediate send.
func (d *PushDispatcher) Dispatch(_ context.Context, delivery DeliveryRow) error {
if d == nil {
return nil
}
d.core.dispatch(delivery.ID)
return nil
}
// Run consumes the dispatch queue and the retry/recovery sweep until ctx is
// canceled.
func (d *PushDispatcher) Run(ctx context.Context) {
d.core.run(ctx)
}
type ApplePushTestResult struct {
AttemptID string
PushDeviceID string
ServerDeviceID string
Outcome string
RelayRequestID string
UpstreamStatus *int
UpstreamReason string
FailureMessage string
}
func (s *System) SendApplePushTest(ctx context.Context, profileID, serverDeviceID string) (*ApplePushTestResult, error) {
if s == nil || s.pushDeviceRepo == nil || s.pushSender == nil {
return nil, ErrPushDeliveryUnavailable
}
if !s.Settings.ApplePushDeliveryEnabled(ctx) {
return nil, ErrPushDeliveryUnavailable
}
if s.Settings.PushRelayAPIKey(ctx) == "" {
return nil, ErrPushDeliveryUnavailable
}
attempt, device, err := s.pushDeviceRepo.EnqueueAppleTestAttempt(ctx, profileID, serverDeviceID)
if err != nil {
return nil, err
}
claimed, err := s.pushDeviceRepo.ClaimPushAttemptByID(ctx, attempt.ID)
if err != nil {
return nil, err
}
if len(claimed) != 1 {
return nil, fmt.Errorf("push test attempt was not claimable")
}
updated := s.pushSender.processAttempt(ctx, claimed[0])
if updated == nil {
updated, _ = s.pushDeviceRepo.GetPushAttempt(ctx, attempt.ID)
}
if updated == nil {
return nil, fmt.Errorf("push test attempt disappeared")
}
result := &ApplePushTestResult{
AttemptID: updated.ID,
PushDeviceID: updated.PushDeviceID,
ServerDeviceID: device.ServerDeviceID,
Outcome: updated.Outcome,
UpstreamStatus: updated.UpstreamStatus,
}
if updated.RelayRequestID != nil {
result.RelayRequestID = *updated.RelayRequestID
}
if updated.UpstreamReason != nil {
result.UpstreamReason = *updated.UpstreamReason
}
if updated.FailureMessage != nil {
result.FailureMessage = *updated.FailureMessage
}
return result, nil
}
+169
View File
@@ -0,0 +1,169 @@
package notifications
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestApplePushDeliverySettings(t *testing.T) {
ctx := context.Background()
settings := NewSettings(nil)
if settings.ApplePushDeliveryEnabled(ctx) {
t.Fatal("ApplePushDeliveryEnabled must default to false")
}
if got := settings.PushRelayURL(ctx); got != DefaultPushRelayURL {
t.Fatalf("PushRelayURL default = %q, want %q", got, DefaultPushRelayURL)
}
settings = NewSettings(mapSettingReader{
SettingApplePushDeliveryEnabled: "true",
SettingPushRelayURL: "https://push.example.test/",
SettingPushRelayAPIKey: " relay-key ",
})
if !settings.ApplePushDeliveryEnabled(ctx) {
t.Fatal("ApplePushDeliveryEnabled = false with setting on")
}
if got := settings.PushRelayURL(ctx); got != "https://push.example.test" {
t.Fatalf("PushRelayURL = %q", got)
}
if got := settings.PushRelayAPIKey(ctx); got != "relay-key" {
t.Fatalf("PushRelayAPIKey was not trimmed")
}
}
func TestPushSenderSendBuildsRelayRequest(t *testing.T) {
token := strings.Repeat("a", 64)
var got struct {
auth string
idempotencyKey string
body pushRelayAppleRequest
}
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got.auth = r.Header.Get("Authorization")
got.idempotencyKey = r.Header.Get("Idempotency-Key")
if r.URL.Path != relayAppleSendPath {
t.Fatalf("path = %q, want %q", r.URL.Path, relayAppleSendPath)
}
if err := json.NewDecoder(r.Body).Decode(&got.body); err != nil {
t.Fatalf("decode request body: %v", err)
}
_ = json.NewEncoder(w).Encode(pushRelayAppleResponse{
RequestID: "relay-request-1",
APNsID: "apns-1",
Status: "accepted",
})
}))
defer server.Close()
settings := NewSettings(mapSettingReader{
SettingPushRelayURL: server.URL,
SettingPushRelayAPIKey: "relay-key",
})
sender := newPushSender(nil, nil, nil, settings)
sender.client = server.Client()
deliveryID := "delivery-1"
result := sender.send(context.Background(), PushDeliveryAttempt{
ID: "attempt-1",
NotificationDeliveryID: &deliveryID,
AttemptNumber: 1,
}, &PushDevice{
APNsEnvironment: APNsEnvironmentSandbox,
APNsTopic: ApplePushTopicSilo,
ServerDeviceID: "server-device-1",
}, token)
if !result.OK || result.RelayRequestID != "relay-request-1" {
t.Fatalf("result = %+v", result)
}
if got.auth != "Bearer relay-key" {
t.Fatalf("Authorization = %q", got.auth)
}
if got.idempotencyKey != "attempt-1:2" {
t.Fatalf("Idempotency-Key = %q", got.idempotencyKey)
}
if got.body.Token != token || got.body.Mode != "private_alert" || got.body.DeliveryID != deliveryID {
t.Fatalf("relay body = %+v", got.body)
}
if got.body.CollapseID == nil || *got.body.CollapseID != deliveryID {
t.Fatalf("collapse_id = %+v, want delivery id", got.body.CollapseID)
}
}
func TestPushSenderSendMapsRelayTerminalAPNsRejection(t *testing.T) {
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnprocessableEntity)
_ = json.NewEncoder(w).Encode(pushRelayErrorResponse{
Error: struct {
Code string `json:"code"`
Message string `json:"message"`
RequestID string `json:"request_id"`
}{
Code: "apns_rejected",
Message: "APNs rejected the notification: BadDeviceToken",
RequestID: "relay-request-2",
},
})
}))
defer server.Close()
settings := NewSettings(mapSettingReader{
SettingPushRelayURL: server.URL,
SettingPushRelayAPIKey: "relay-key",
})
sender := newPushSender(nil, nil, nil, settings)
sender.client = server.Client()
result := sender.send(context.Background(), PushDeliveryAttempt{ID: "attempt-1"}, &PushDevice{
APNsEnvironment: APNsEnvironmentSandbox,
APNsTopic: ApplePushTopicSilo,
ServerDeviceID: "server-device-1",
}, strings.Repeat("a", 64))
if result.OK || !result.TerminalDevice || result.HTTPStatus != http.StatusUnprocessableEntity {
t.Fatalf("terminal result = %+v", result)
}
if result.UpstreamReason != "apns_rejected" || result.RelayRequestID != "relay-request-2" {
t.Fatalf("terminal diagnostic = %+v", result)
}
}
func TestPushSenderSendMapsRelayRetryAfter(t *testing.T) {
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Retry-After", "30")
w.WriteHeader(http.StatusTooManyRequests)
_ = json.NewEncoder(w).Encode(pushRelayErrorResponse{
Error: struct {
Code string `json:"code"`
Message string `json:"message"`
RequestID string `json:"request_id"`
}{
Code: "upstream_rate_limited",
Message: "APNs upstream rate limited the request",
RequestID: "relay-request-3",
},
})
}))
defer server.Close()
settings := NewSettings(mapSettingReader{
SettingPushRelayURL: server.URL,
SettingPushRelayAPIKey: "relay-key",
})
sender := newPushSender(nil, nil, nil, settings)
sender.client = server.Client()
result := sender.send(context.Background(), PushDeliveryAttempt{ID: "attempt-1"}, &PushDevice{
APNsEnvironment: APNsEnvironmentSandbox,
APNsTopic: ApplePushTopicSilo,
ServerDeviceID: "server-device-1",
}, strings.Repeat("a", 64))
if result.OK || result.TerminalDevice || result.RetryAfter == 0 {
t.Fatalf("retryable result = %+v", result)
}
}
+37
View File
@@ -51,6 +51,11 @@ const (
SettingServerChannelsBatchSeconds = "notifications.server_channels.batch_seconds"
SettingServerChannelMentionRequesters = "notifications.server_channels.mention_requesters"
SettingApplePushDeliveryEnabled = "notifications.apple_push_delivery_enabled"
SettingPushRelayURL = "notifications.push_relay_url"
SettingPushRelayDeploymentID = "notifications.push_relay_deployment_id"
SettingPushRelayAPIKey = "notifications.push_relay_api_key"
// Discord application credentials live under the discord.* namespace
// (admin-configured, alongside email.smtp_*). The secret and bot token
// are registered in catalog.SensitiveSettingKeys and encrypted at rest.
@@ -78,6 +83,9 @@ const (
// already passed, and the window is what keeps those visible.
defaultServerChannelsBatchSeconds = 300
minServerChannelsBatchSeconds = 120
// DefaultPushRelayURL is the public Silo relay origin used when no
// notifications.push_relay_url override is stored.
DefaultPushRelayURL = "https://push.siloserver.org"
settingsCacheTTL = 15 * time.Second
)
@@ -347,6 +355,35 @@ func (s *Settings) ServerChannelsBatchWindow(ctx context.Context) time.Duration
defaultServerChannelsBatchSeconds, minServerChannelsBatchSeconds, 3600)) * time.Second
}
// ApplePushDeliveryEnabled gates relay sends and the capability endpoint's
// apple_push availability, mirroring how web push advertises itself. The
// device registration endpoint stays available independently so clients that
// already hold tokens keep them fresh across admin toggles.
func (s *Settings) ApplePushDeliveryEnabled(ctx context.Context) bool {
return s.boolSetting(ctx, SettingApplePushDeliveryEnabled, false)
}
// PushRelayURL is the public Silo relay origin.
func (s *Settings) PushRelayURL(ctx context.Context) string {
value := strings.TrimRight(strings.TrimSpace(s.raw(ctx, SettingPushRelayURL)), "/")
if value == "" {
return DefaultPushRelayURL
}
return value
}
// PushRelayAPIKey is the bearer credential for the Silo relay.
func (s *Settings) PushRelayAPIKey(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingPushRelayAPIKey))
}
// PushRelayDeploymentID is the relay account identifier returned during
// self-registration. The dispatcher does not need it; registration uses it for
// subsequent key rotations.
func (s *Settings) PushRelayDeploymentID(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingPushRelayDeploymentID))
}
// DiscordClientID is the Discord application's OAuth2 client ID.
func (s *Settings) DiscordClientID(ctx context.Context) string {
return strings.TrimSpace(s.raw(ctx, SettingDiscordClientID))
+34
View File
@@ -54,6 +54,9 @@ type System struct {
// WebPush is nil when the settings store is not writable (VAPID keys
// could not be provisioned).
WebPush *WebPushService
// PushDevices is nil when no at-rest cipher is configured; APNs tokens are
// credentials and must not be stored in plaintext.
PushDevices *PushDeviceService
// EmailPrefs is nil when no mail sender was provided.
EmailPrefs *EmailPrefsRepository
// DiscordPrefs holds Discord DM link + mode state; the channel only
@@ -73,6 +76,9 @@ type System struct {
webhookRetry *WebhookRetryWorker
webPushRepo *WebPushRepository
webPushDispatcher *WebPushDispatcher
pushDeviceRepo *PushDeviceRepository
pushDispatcher *PushDispatcher
pushSender *pushSender
// serverChannelWorker sweeps release_events into admin broadcast posts;
// nil without the at-rest cipher.
serverChannelWorker *serverChannelWorker
@@ -119,6 +125,10 @@ func NewSystem(
var webhookDispatcher *WebhookDispatcher
var webhookRetry *WebhookRetryWorker
var sender *webhookSender
var pushDeviceService *PushDeviceService
var pushDeviceRepo *PushDeviceRepository
var pushSenderInst *pushSender
var pushDispatcher *PushDispatcher
if cipher != nil {
webhookRepo = NewWebhookRepository(pool)
sender = newWebhookSender(webhookRepo, deliveries, cipher, settings)
@@ -126,6 +136,11 @@ func NewSystem(
webhookDispatcher = newWebhookDispatcher(sender)
webhookRetry = newWebhookRetryWorker(sender)
dispatchers = append(dispatchers, webhookDispatcher)
pushDeviceRepo = NewPushDeviceRepository(pool)
pushDeviceService = NewPushDeviceService(pushDeviceRepo, cipher)
pushSenderInst = newPushSender(pushDeviceRepo, deliveries, cipher, settings)
pushDispatcher = newPushDispatcher(pushSenderInst)
dispatchers = append(dispatchers, pushDispatcher)
}
// Admin server channels (broadcast destinations) share the cipher
@@ -187,6 +202,9 @@ func NewSystem(
if webPushRepo != nil {
fanout.SetWebPushOutbox(webPushRepo)
}
if pushDeviceRepo != nil {
fanout.SetPushOutbox(pushDeviceRepo)
}
detector := NewAvailabilityDetector(releases, settings)
detector.SetFanoutNudge(func() {
fanout.Nudge()
@@ -207,6 +225,7 @@ func NewSystem(
Webhooks: webhookService,
ServerChannels: serverChannelService,
WebPush: webPushService,
PushDevices: pushDeviceService,
EmailPrefs: emailPrefs,
DiscordPrefs: discordPrefs,
mailSender: mailSender,
@@ -218,6 +237,9 @@ func NewSystem(
webhookRetry: webhookRetry,
webPushRepo: webPushRepo,
webPushDispatcher: webPushDispatcher,
pushDeviceRepo: pushDeviceRepo,
pushDispatcher: pushDispatcher,
pushSender: pushSenderInst,
serverChannelWorker: serverChannelSweep,
dispatcher: multiDispatcher,
pool: pool,
@@ -354,6 +376,13 @@ func (s *System) Start(ctx context.Context) {
}
}()
}
if s.pushDispatcher != nil {
s.wg.Add(1)
go func() {
defer s.wg.Done()
s.pushDispatcher.Run(ctx)
}()
}
}
// Wait blocks until the background loops exit (after their context is
@@ -390,6 +419,11 @@ func (s *System) PurgeProfile(ctx context.Context, profileID string) error {
return fmt.Errorf("purge web push subscriptions: %w", err)
}
}
if s.pushDeviceRepo != nil {
if err := s.pushDeviceRepo.DeleteAllForProfile(ctx, profileID); err != nil {
return fmt.Errorf("purge push devices: %w", err)
}
}
if s.EmailPrefs != nil {
if err := s.EmailPrefs.DeleteForProfile(ctx, profileID); err != nil {
return fmt.Errorf("purge email prefs: %w", err)
+15 -82
View File
@@ -3,7 +3,6 @@ package notifications
import (
"context"
"log/slog"
"sync"
"time"
)
@@ -15,23 +14,20 @@ const (
)
// WebhookDispatcher implements the channel Dispatcher interface for outbound
// webhooks. Dispatch never blocks the fanout loop on destination HTTP: it
// hands the delivery ID to a bounded worker pool that claims the delivery's
// pending outbox attempts and sends them. A full queue simply drops the
// hand-off — the durable `pending` rows are picked up by the retry worker's
// outbox recovery sweep, so delivery is delayed, never lost.
// webhooks on top of the shared channelDispatcher core. Retry/recovery runs in
// the standalone WebhookRetryWorker.
type WebhookDispatcher struct {
sender *webhookSender
queue chan string
logger *slog.Logger
core channelDispatcher[DeliveryAttempt]
}
func newWebhookDispatcher(sender *webhookSender) *WebhookDispatcher {
return &WebhookDispatcher{
sender: sender,
queue: make(chan string, webhookDispatchQueue),
logger: slog.Default().With("component", "notifications.webhooks.dispatch"),
}
return &WebhookDispatcher{core: channelDispatcher[DeliveryAttempt]{
channel: "webhook",
queue: make(chan string, webhookDispatchQueue),
logger: slog.Default().With("component", "notifications.webhooks.dispatch"),
claimPending: sender.webhooks.ClaimPendingForDelivery,
process: sender.processAttempt,
}}
}
// Dispatch queues the delivery's webhook attempts for immediate send.
@@ -44,50 +40,14 @@ func (d *WebhookDispatcher) Dispatch(_ context.Context, delivery DeliveryRow) er
// webhook, or a broken webhook would loop forever.
return nil
}
select {
case d.queue <- delivery.ID:
default:
d.logger.Warn("webhook dispatch queue full; deferring to retry worker",
"delivery_id", delivery.ID)
}
d.core.dispatch(delivery.ID)
return nil
}
// Run consumes the dispatch queue with a bounded worker pool until ctx is
// canceled. One slow destination cannot block other deliveries.
func (d *WebhookDispatcher) Run(ctx context.Context) {
var wg sync.WaitGroup
for range webhookDispatchWorkers {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case deliveryID := <-d.queue:
d.processDelivery(ctx, deliveryID)
}
}
}()
}
wg.Wait()
}
func (d *WebhookDispatcher) processDelivery(ctx context.Context, deliveryID string) {
attempts, err := d.sender.webhooks.ClaimPendingForDelivery(ctx, deliveryID)
if err != nil {
if ctx.Err() == nil {
d.logger.Warn("webhook attempt claim failed", "delivery_id", deliveryID, "error", err)
}
return
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
d.sender.processAttempt(ctx, attempt)
}
d.core.run(ctx)
}
// WebhookRetryWorker drains due retries and recovers stale pending outbox
@@ -107,34 +67,7 @@ func newWebhookRetryWorker(sender *webhookSender) *WebhookRetryWorker {
// Run polls for due attempts until ctx is canceled.
func (w *WebhookRetryWorker) Run(ctx context.Context) {
ticker := time.NewTicker(webhookRetryInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
if !w.sender.settings.WebhooksEnabled(ctx) {
continue
}
for {
attempts, err := w.sender.webhooks.ClaimDue(ctx, webhookRetryClaimLimit)
if err != nil {
if ctx.Err() == nil {
w.logger.Warn("webhook retry claim failed", "error", err)
}
break
}
if len(attempts) == 0 {
break
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
w.sender.processAttempt(ctx, attempt)
}
}
}
runRetrySweep(ctx, "webhook", w.logger,
w.sender.settings.WebhooksEnabled, w.sender.webhooks.ClaimDue,
webhookRetryClaimLimit, w.sender.processAttempt)
}
+37 -2
View File
@@ -18,10 +18,10 @@ func TestBuildWebPushPayload(t *testing.T) {
if err := json.Unmarshal(raw, &payload); err != nil {
t.Fatal(err)
}
if payload.Title != "New episode of Severance" {
if payload.Title != "The latest episode of Severance S02E01 just dropped!" {
t.Fatalf("unexpected title %q", payload.Title)
}
if payload.Body != "S2E1 — Hello, Ms. Cobel" {
if payload.Body != "Hello, Ms. Cobel" {
t.Fatalf("unexpected body %q", payload.Body)
}
if payload.URL != "/item/episode-456" {
@@ -83,6 +83,41 @@ func TestBuildWebPushPayload(t *testing.T) {
})
}
func TestBuildNotificationDisplay(t *testing.T) {
t.Run("episode available", func(t *testing.T) {
display := BuildNotificationDisplay(webhookTestRow())
if display.DeliveryID != "01DELIVERY" ||
display.Title != "The latest episode of Severance S02E01 just dropped!" ||
display.Body != "Hello, Ms. Cobel" ||
display.ThreadID != "series:series-123" ||
display.Category != "episode_available" ||
display.URL != "/item/episode-456" {
t.Fatalf("display = %+v", display)
}
})
t.Run("request declined", func(t *testing.T) {
row := requestDeclinedTestRow()
display := BuildNotificationDisplay(row)
if display.Title != "Dune was declined" ||
display.Body != "Reason: Already available in 4K" ||
display.ThreadID != "request:01REQ" ||
display.Category != "request_declined" ||
display.URL != "/notifications" {
t.Fatalf("display = %+v", display)
}
})
t.Run("unknown type", func(t *testing.T) {
display := BuildNotificationDisplay(DeliveryRow{Delivery: Delivery{ID: "01X", Type: "future.type"}})
if display.Title != genericNotificationTitle ||
display.Category != "future_type" ||
display.URL != "/notifications" {
t.Fatalf("display = %+v", display)
}
})
}
func TestWebPushRetrySchedule(t *testing.T) {
total := time.Duration(0)
for attempt := 1; attempt < webPushMaxAttempts; attempt++ {
+19 -140
View File
@@ -7,7 +7,6 @@ import (
"io"
"log/slog"
"net/http"
"sync"
"time"
webpush "github.com/SherClockHolmes/webpush-go"
@@ -51,70 +50,19 @@ type webPushPayload struct {
// buildWebPushPayload renders a delivery for the service worker.
func buildWebPushPayload(row DeliveryRow, posterURL string) ([]byte, error) {
display := BuildNotificationDisplay(row)
payload := webPushPayload{
Title: "Silo",
URL: "/notifications",
Title: display.Title,
Body: display.Body,
URL: display.URL,
Tag: row.ID,
DeliveryID: row.ID,
}
switch row.Type {
case DeliveryTypeEpisodeAvailable:
if row.SeriesTitle != "" {
payload.Title = "New episode of " + row.SeriesTitle
} else {
payload.Title = "New episode available"
}
var code string
if row.SeasonNumber != nil && row.EpisodeNumber != nil {
code = fmt.Sprintf("S%dE%d", *row.SeasonNumber, *row.EpisodeNumber)
}
switch {
case code != "" && row.EpisodeTitle != "":
payload.Body = code + " — " + row.EpisodeTitle
case code != "":
payload.Body = code
default:
payload.Body = row.EpisodeTitle
}
if row.EpisodeID != nil {
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 DeliveryTypeRequestApproved:
flags := parseRequestFlags(row.ReasonFlags)
payload.Title = "Your request was approved"
if flags.Title != "" {
payload.Title = flags.Title + " was approved"
}
payload.Body = "Your media request was approved."
case DeliveryTypeRequestDeclined:
flags := parseRequestFlags(row.ReasonFlags)
payload.Title = "Your request was declined"
if flags.Title != "" {
payload.Title = flags.Title + " was declined"
}
payload.Body = "Your media request was declined."
if flags.Reason != "" {
payload.Body = "Reason: " + flags.Reason
}
case DeliveryTypeWebhookAutoDisabled:
payload.Title = "A webhook stopped working"
payload.Body = "Open notification settings to fix it."
payload.URL = "/settings/notifications"
default:
// Unknown types render generically; the inbox has the details.
payload.Title = genericNotificationTitle
}
return json.Marshal(payload)
}
@@ -280,21 +228,23 @@ func (s *webPushSender) send(ctx context.Context, sub *WebPushSubscription, mess
return resp.StatusCode, retryAfter, nil
}
// WebPushDispatcher implements the channel Dispatcher interface: it hands
// delivery IDs to a bounded worker pool that claims and sends the pending
// outbox attempts. A full queue defers to the retry worker's recovery sweep.
// WebPushDispatcher implements the channel Dispatcher interface on top of the
// shared channelDispatcher core, with the retry/recovery sweep integrated.
type WebPushDispatcher struct {
sender *webPushSender
queue chan string
logger *slog.Logger
core channelDispatcher[DeliveryAttempt]
}
func newWebPushDispatcher(sender *webPushSender) *WebPushDispatcher {
return &WebPushDispatcher{
sender: sender,
queue: make(chan string, webhookDispatchQueue),
logger: slog.Default().With("component", "notifications.webpush.dispatch"),
}
return &WebPushDispatcher{core: channelDispatcher[DeliveryAttempt]{
channel: "web push",
queue: make(chan string, webhookDispatchQueue),
logger: slog.Default().With("component", "notifications.webpush.dispatch"),
claimPending: sender.subscriptions.ClaimPendingForDelivery,
process: sender.processAttempt,
enabled: sender.settings.WebPushEnabled,
claimDue: sender.subscriptions.ClaimDue,
claimLimit: webhookRetryClaimLimit,
}}
}
// Dispatch queues the delivery's web push attempts for immediate send.
@@ -302,83 +252,12 @@ func (d *WebPushDispatcher) Dispatch(_ context.Context, delivery DeliveryRow) er
if d == nil {
return nil
}
select {
case d.queue <- delivery.ID:
default:
d.logger.Warn("web push dispatch queue full; deferring to retry worker",
"delivery_id", delivery.ID)
}
d.core.dispatch(delivery.ID)
return nil
}
// Run consumes the dispatch queue and the retry/recovery sweep until ctx is
// canceled.
func (d *WebPushDispatcher) Run(ctx context.Context) {
var wg sync.WaitGroup
for range webhookDispatchWorkers {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case deliveryID := <-d.queue:
d.processDelivery(ctx, deliveryID)
}
}
}()
}
wg.Add(1)
go func() {
defer wg.Done()
ticker := time.NewTicker(webhookRetryInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
if !d.sender.settings.WebPushEnabled(ctx) {
continue
}
for {
attempts, err := d.sender.subscriptions.ClaimDue(ctx, webhookRetryClaimLimit)
if err != nil {
if ctx.Err() == nil {
d.logger.Warn("web push retry claim failed", "error", err)
}
break
}
if len(attempts) == 0 {
break
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
d.sender.processAttempt(ctx, attempt)
}
}
}
}()
wg.Wait()
}
func (d *WebPushDispatcher) processDelivery(ctx context.Context, deliveryID string) {
attempts, err := d.sender.subscriptions.ClaimPendingForDelivery(ctx, deliveryID)
if err != nil {
if ctx.Err() == nil {
d.logger.Warn("web push attempt claim failed", "delivery_id", deliveryID, "error", err)
}
return
}
for _, attempt := range attempts {
if ctx.Err() != nil {
return
}
d.sender.processAttempt(ctx, attempt)
}
d.core.run(ctx)
}
+60
View File
@@ -0,0 +1,60 @@
package migrations
import (
"strings"
"testing"
)
func TestPushDevicesMigrationContract(t *testing.T) {
migrationBytes, err := FS.ReadFile("sql/20260701143000_push_devices.sql")
if err != nil {
t.Fatalf("read migration: %v", err)
}
migration := string(migrationBytes)
for _, want := range []string{
"CREATE TABLE public.push_devices",
"CONSTRAINT push_devices_profile_device_platform_key UNIQUE (profile_id, device_id, platform)",
"CONSTRAINT push_devices_server_device_id_key UNIQUE (server_device_id)",
"CONSTRAINT push_devices_platform_check CHECK (platform IN ('apple'))",
"CONSTRAINT push_devices_provider_check CHECK (provider IN ('silo_relay'))",
"CONSTRAINT push_devices_apns_environment_check CHECK (apns_environment IN ('production', 'sandbox'))",
"CONSTRAINT push_devices_push_mode_check CHECK (push_mode IN ('off', 'in_app_only', 'private_push'))",
"CONSTRAINT push_devices_apple_fields_check CHECK",
"CREATE INDEX push_devices_profile_enabled_idx ON public.push_devices (profile_id) WHERE enabled",
} {
if !strings.Contains(migration, want) {
t.Fatalf("migration missing %q", want)
}
}
if strings.Contains(migration, "REFERENCES profiles") {
t.Fatal("push_devices migration must not add a profile foreign key")
}
}
func TestPushDeliveryAttemptsMigrationContract(t *testing.T) {
migrationBytes, err := FS.ReadFile("sql/20260701170000_push_delivery_attempts.sql")
if err != nil {
t.Fatalf("read migration: %v", err)
}
migration := string(migrationBytes)
for _, want := range []string{
"CREATE TABLE public.push_delivery_attempts",
"notification_delivery_id text REFERENCES public.notification_deliveries(id) ON DELETE CASCADE",
"push_device_id text NOT NULL REFERENCES public.push_devices(id) ON DELETE CASCADE",
"CONSTRAINT push_delivery_attempts_trigger_check CHECK (trigger_type IN ('delivery', 'test'))",
"CONSTRAINT push_delivery_attempts_delivery_required_check CHECK",
"CONSTRAINT push_delivery_attempts_provider_check CHECK (provider IN ('silo_relay'))",
"CONSTRAINT push_delivery_attempts_platform_check CHECK (platform IN ('apple'))",
"CONSTRAINT push_delivery_attempts_outcome_check CHECK (outcome IN ('pending', 'delivered', 'retrying', 'failed'))",
"CREATE UNIQUE INDEX push_delivery_attempts_delivery_unique",
"CREATE INDEX push_delivery_attempts_retry_idx",
"CREATE INDEX push_delivery_attempts_device_history_idx",
} {
if !strings.Contains(migration, want) {
t.Fatalf("migration missing %q", want)
}
}
if strings.Contains(migration, "REFERENCES profiles") {
t.Fatal("push delivery attempts migration must not add a profile foreign key")
}
}
@@ -0,0 +1,46 @@
-- +goose Up
-- +goose StatementBegin
CREATE TABLE public.push_devices (
id text PRIMARY KEY,
user_id integer NOT NULL,
profile_id text NOT NULL,
device_id varchar(128) NOT NULL,
platform text NOT NULL,
provider text NOT NULL,
apns_environment text,
apns_topic text,
apns_token_ciphertext text,
apns_token_hash text,
server_device_id text NOT NULL,
push_mode text NOT NULL DEFAULT 'private_push',
enabled boolean NOT NULL DEFAULT true,
last_seen_at timestamptz,
last_success_at timestamptz,
last_failure_at timestamptz,
last_failure_code text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
CONSTRAINT push_devices_profile_device_platform_key UNIQUE (profile_id, device_id, platform),
CONSTRAINT push_devices_server_device_id_key UNIQUE (server_device_id),
CONSTRAINT push_devices_platform_check CHECK (platform IN ('apple')),
CONSTRAINT push_devices_provider_check CHECK (provider IN ('silo_relay')),
CONSTRAINT push_devices_apns_environment_check CHECK (apns_environment IN ('production', 'sandbox')),
CONSTRAINT push_devices_push_mode_check CHECK (push_mode IN ('off', 'in_app_only', 'private_push')),
CONSTRAINT push_devices_apple_fields_check CHECK (
platform = 'apple'
AND apns_environment IS NOT NULL
AND apns_topic IS NOT NULL
AND apns_token_ciphertext IS NOT NULL
AND apns_token_hash IS NOT NULL
)
);
CREATE INDEX push_devices_profile_idx ON public.push_devices (profile_id);
CREATE INDEX push_devices_profile_enabled_idx ON public.push_devices (profile_id) WHERE enabled;
CREATE INDEX push_devices_apns_token_hash_idx ON public.push_devices (apns_token_hash);
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
DROP TABLE IF EXISTS public.push_devices;
-- +goose StatementEnd
@@ -0,0 +1,45 @@
-- +goose Up
-- +goose StatementBegin
CREATE TABLE public.push_delivery_attempts (
id text PRIMARY KEY,
notification_delivery_id text REFERENCES public.notification_deliveries(id) ON DELETE CASCADE,
push_device_id text NOT NULL REFERENCES public.push_devices(id) ON DELETE CASCADE,
trigger_type text NOT NULL DEFAULT 'delivery',
provider text NOT NULL,
platform text NOT NULL,
attempt_number integer NOT NULL DEFAULT 0,
attempted_at timestamptz NOT NULL DEFAULT now(),
next_retry_at timestamptz,
outcome text NOT NULL DEFAULT 'pending',
relay_request_id text,
upstream_status integer,
upstream_reason varchar(256),
failure_message varchar(256),
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
CONSTRAINT push_delivery_attempts_trigger_check CHECK (trigger_type IN ('delivery', 'test')),
CONSTRAINT push_delivery_attempts_delivery_required_check CHECK (
(trigger_type = 'delivery' AND notification_delivery_id IS NOT NULL)
OR trigger_type = 'test'
),
CONSTRAINT push_delivery_attempts_provider_check CHECK (provider IN ('silo_relay')),
CONSTRAINT push_delivery_attempts_platform_check CHECK (platform IN ('apple')),
CONSTRAINT push_delivery_attempts_outcome_check CHECK (outcome IN ('pending', 'delivered', 'retrying', 'failed'))
);
CREATE UNIQUE INDEX push_delivery_attempts_delivery_unique
ON public.push_delivery_attempts (push_device_id, notification_delivery_id, attempt_number)
WHERE notification_delivery_id IS NOT NULL;
CREATE INDEX push_delivery_attempts_delivery_idx
ON public.push_delivery_attempts (notification_delivery_id)
WHERE notification_delivery_id IS NOT NULL;
CREATE INDEX push_delivery_attempts_retry_idx
ON public.push_delivery_attempts (outcome, next_retry_at);
CREATE INDEX push_delivery_attempts_device_history_idx
ON public.push_delivery_attempts (push_device_id, attempted_at DESC);
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
DROP TABLE IF EXISTS public.push_delivery_attempts;
-- +goose StatementEnd
+13 -1
View File
@@ -324,10 +324,17 @@ export const ADMIN_SETTINGS_GROUPS: AdminSettingsSearchGroup[] = [
id: "notifications",
label: "Notifications",
description:
"Server notification channels, release events, Discord, web push, and webhooks.",
"Server notification channels, release events, Silo Push Relay, Discord, web push, and webhooks.",
keywords: [
"release events",
"new episode",
"silo push relay",
"mobile push",
"apple push",
"android push",
"apns",
"push relay",
"privacy disclosure",
"discord",
"browser push",
"web push",
@@ -342,6 +349,11 @@ export const ADMIN_SETTINGS_GROUPS: AdminSettingsSearchGroup[] = [
"Delivery Channels",
"In-App",
"Web Push",
"Silo Push Relay",
"Relay URL",
"Deployment ID",
"Register Relay",
"Privacy Disclosure",
"Email",
"Allow Per-Episode Email",
"Digest Hour",
@@ -1,3 +1,4 @@
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { fireEvent, render, screen } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { renderToStaticMarkup } from "react-dom/server";
@@ -13,10 +14,26 @@ vi.mock("@/hooks/useSettingsForm", () => ({
}));
function renderLayout(search = "") {
const client = new QueryClient({ defaultOptions: { queries: { retry: false } } });
return renderToStaticMarkup(
<MemoryRouter initialEntries={[`/admin/settings${search}`]}>
<AdminSettingsLayout />
</MemoryRouter>,
<QueryClientProvider client={client}>
<MemoryRouter initialEntries={[`/admin/settings${search}`]}>
<AdminSettingsLayout />
</MemoryRouter>
</QueryClientProvider>,
);
}
function renderInteractiveLayout(search = "") {
const client = new QueryClient({ defaultOptions: { queries: { retry: false } } });
return render(
<QueryClientProvider client={client}>
<MemoryRouter initialEntries={[`/admin/settings${search}`]}>
<AdminSettingsLayout />
</MemoryRouter>
</QueryClientProvider>,
);
}
@@ -71,11 +88,7 @@ describe("AdminSettingsLayout", () => {
});
it("filters admin settings sections from the search box", async () => {
render(
<MemoryRouter initialEntries={["/admin/settings"]}>
<AdminSettingsLayout />
</MemoryRouter>,
);
renderInteractiveLayout();
await userEvent.type(screen.getByRole("searchbox", { name: "Search settings" }), "redis");
@@ -85,11 +98,7 @@ describe("AdminSettingsLayout", () => {
});
it("matches individual admin setting labels", async () => {
render(
<MemoryRouter initialEntries={["/admin/settings"]}>
<AdminSettingsLayout />
</MemoryRouter>,
);
renderInteractiveLayout();
await userEvent.type(
screen.getByRole("searchbox", { name: "Search settings" }),
@@ -101,11 +110,7 @@ describe("AdminSettingsLayout", () => {
});
it("focuses admin settings search with Cmd+K", () => {
render(
<MemoryRouter initialEntries={["/admin/settings"]}>
<AdminSettingsLayout />
</MemoryRouter>,
);
renderInteractiveLayout();
const searchBox = screen.getByRole("searchbox", { name: "Search settings" });
fireEvent.keyDown(document, { key: "k", metaKey: true });
@@ -0,0 +1,115 @@
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { render, screen } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { renderToStaticMarkup } from "react-dom/server";
import { MemoryRouter } from "react-router";
import { describe, expect, it, vi } from "vitest";
import NotificationsAdminSettings from "./NotificationsAdminSettings";
const useSettingsFormMock = vi.fn();
vi.mock("@/hooks/useSettingsForm", () => ({
useSettingsForm: (...args: unknown[]) => useSettingsFormMock(...args),
}));
vi.mock("@/hooks/queries/admin/serverNotificationChannels", () => ({
useServerNotificationChannels: () => ({ data: [] }),
}));
function makeForm() {
return {
isLoading: false,
getValue: (key: string) => {
switch (key) {
case "notifications.release_events_enabled":
case "notifications.fanout_enabled":
case "notifications.ui_enabled":
case "notifications.web_push_enabled":
case "notifications.apple_push_delivery_enabled":
return "true";
case "notifications.push_relay_url":
return "https://push.siloserver.org";
case "notifications.push_relay_deployment_id":
return "01DEPLOYMENT";
default:
return "";
}
},
setValue: vi.fn(),
dirtyCount: 0,
dirtyKeys: [],
isDirty: vi.fn(() => false),
save: vi.fn(),
discard: vi.fn(),
isSaving: false,
restartRequired: false,
sensitiveConfigured: ["notifications.push_relay_api_key"],
sensitiveManagedByEnv: [],
buildConnectionCheckRequest: vi.fn(),
};
}
function renderPage() {
const client = new QueryClient({ defaultOptions: { queries: { retry: false } } });
return (
<QueryClientProvider client={client}>
<MemoryRouter initialEntries={["/admin/settings?tab=notifications"]}>
<NotificationsAdminSettings />
</MemoryRouter>
</QueryClientProvider>
);
}
function renderStaticPage() {
return renderToStaticMarkup(renderPage());
}
describe("NotificationsAdminSettings", () => {
it("registers Silo Push Relay settings with the shared settings form", () => {
useSettingsFormMock.mockReturnValue(makeForm());
renderStaticPage();
expect(useSettingsFormMock).toHaveBeenCalledWith({
keys: expect.arrayContaining([
"notifications.apple_push_delivery_enabled",
"notifications.push_relay_deployment_id",
]),
});
const firstCall = useSettingsFormMock.mock.calls[0];
if (!firstCall) {
throw new Error("useSettingsForm was not called");
}
const [options] = firstCall as [{ keys: string[] }];
expect(options.keys).not.toContain("notifications.push_relay_api_key");
// The relay URL is persisted only via the registration endpoint, never
// through the settings form.
expect(options.keys).not.toContain("notifications.push_relay_url");
});
it("shows the Silo Push Relay channel status", async () => {
useSettingsFormMock.mockReturnValue(makeForm());
render(renderPage());
expect(screen.getByText("Silo Push Relay")).toBeInTheDocument();
expect(screen.getByText(/Mobile push delivery through Silo's relay/)).toBeInTheDocument();
expect(screen.getByText(/Android support will use the same relay/)).toBeInTheDocument();
expect(screen.getByText("Relay configured")).toBeInTheDocument();
await userEvent.click(screen.getByRole("button", { name: /Silo Push Relay/ }));
expect(screen.getByText("Privacy disclosure")).toBeInTheDocument();
expect(screen.getByText(/content-free request to Silo's push relay/)).toBeInTheDocument();
expect(screen.getByText(/does not receive notification titles/)).toBeInTheDocument();
expect(screen.getByText(/fetches private content directly/)).toBeInTheDocument();
expect(screen.getByText("Deployment ID")).toBeInTheDocument();
expect(screen.getByText("Rotate relay key")).toBeInTheDocument();
expect(screen.queryByText("Relay API Key")).not.toBeInTheDocument();
expect(screen.queryByText("Smoke Test Profile ID")).not.toBeInTheDocument();
expect(screen.queryByText("Server Device ID")).not.toBeInTheDocument();
expect(screen.queryByText("Send test push")).not.toBeInTheDocument();
});
});
@@ -1,4 +1,5 @@
import { useId, useMemo, useState } from "react";
import { useQueryClient } from "@tanstack/react-query";
import {
Bell,
BookOpen,
@@ -9,10 +10,12 @@ import {
Copy,
ExternalLink,
Inbox,
KeyRound,
Loader2,
Mail,
Megaphone,
MonitorSmartphone,
RadioTower,
Rss,
Send,
TriangleAlert,
@@ -26,6 +29,7 @@ import { api } from "@/api/client";
import { Button } from "@/components/ui/button";
import { Skeleton } from "@/components/ui/skeleton";
import { Switch } from "@/components/ui/switch";
import { adminKeys } from "@/hooks/queries/keys";
import { useServerNotificationChannels } from "@/hooks/queries/admin/serverNotificationChannels";
import { useSettingsForm } from "@/hooks/useSettingsForm";
import { cn } from "@/lib/utils";
@@ -40,6 +44,11 @@ const KEYS = [
"notifications.ui_enabled",
"notifications.webhooks_enabled",
"notifications.web_push_enabled",
"notifications.apple_push_delivery_enabled",
// notifications.push_relay_url is intentionally absent: the server rejects
// direct writes; the relay URL is persisted by the registration endpoint
// together with the deployment id and API key.
"notifications.push_relay_deployment_id",
"notifications.fanout.settle_seconds",
"notifications.fanout.max_series_burst",
"notifications.fanout.max_event_age_hours",
@@ -71,6 +80,17 @@ interface DiscordTestResult {
message?: string;
}
interface AppleRelayRegisterResult {
relay_url: string;
deployment_id: string;
key_prefix: string;
api_key_configured: boolean;
relay_request_id?: string;
apns_topics?: string[];
}
const DEFAULT_PUSH_RELAY_URL = "https://push.siloserver.org";
/**
* Invite link for adding the bot to a Discord server. Membership alone is
* enough to DM, so no permissions are requested.
@@ -407,9 +427,120 @@ function TestDiscordRow({ unsaved }: { unsaved: boolean }) {
);
}
function RegisterRelayRow({
relayURL,
deploymentID,
urlEdited,
onRegistered,
}: {
relayURL: string;
deploymentID: string;
urlEdited: boolean;
onRegistered: () => void;
}) {
const queryClient = useQueryClient();
const [pending, setPending] = useState(false);
const [result, setResult] = useState<AppleRelayRegisterResult | null>(null);
const configured = deploymentID.trim() !== "";
const registerRelay = async () => {
if (pending) return;
setPending(true);
setResult(null);
try {
const response = await api<AppleRelayRegisterResult>(
"/admin/notifications/push/relay/register",
{
method: "POST",
body: JSON.stringify({
relay_url: relayURL,
}),
},
);
setResult(response);
await Promise.all([
queryClient.invalidateQueries({ queryKey: adminKeys.serverSettings() }),
queryClient.invalidateQueries({
queryKey: [...adminKeys.serverSettings(), "sensitive-status"] as const,
}),
]);
onRegistered();
toast.success("Push relay registered");
} catch (error) {
toast.error(error instanceof Error ? error.message : "Relay registration failed");
} finally {
setPending(false);
}
};
return (
<div className="space-y-3 py-3">
<SettingField
label="Deployment ID"
hint="Opaque relay account for this Silo server; created during registration"
type="text"
value={deploymentID}
onChange={() => {}}
disabled
/>
<div className="flex flex-wrap items-center gap-2 py-2">
<Button variant="outline" size="sm" disabled={pending} onClick={() => void registerRelay()}>
{pending ? (
<Loader2 className="mr-1.5 h-3.5 w-3.5 animate-spin" />
) : (
<KeyRound className="mr-1.5 h-3.5 w-3.5" />
)}
{configured ? "Rotate relay key" : "Register relay"}
</Button>
</div>
{urlEdited && (
<div className="text-muted-foreground text-xs">
The relay URL change is applied when you register; credentials are stored immediately.
</div>
)}
{result && (
<div className="text-xs text-emerald-500">
Registered {result.deployment_id}
{result.key_prefix ? ` — key ${result.key_prefix}` : ""}
{result.relay_request_id ? ` — relay ${result.relay_request_id}` : ""}
</div>
)}
</div>
);
}
function ApplePushPrivacyDisclosure() {
return (
<div className="space-y-2 py-3">
<div className="text-sm font-medium">Privacy disclosure</div>
<div className="text-muted-foreground space-y-2 text-xs leading-relaxed">
<p>
If you enable push notifications, your Silo Server sends a content-free request to Silo's
push relay so Silo can deliver notifications through Apple Push Notification service.
</p>
<p>
The relay does not receive notification titles, message bodies, media names, user names,
profile names, or your server URL. It does process technical metadata needed to deliver
and operate the service, including an opaque deployment identifier, push delivery timing,
request status, app topic, the IP address your self-hosted Silo Server uses to contact the
relay, and a hashed device push token. Apple may also process standard APNs delivery
metadata.
</p>
<p>
Push notifications are generic; the app fetches private content directly from your Silo
Server after receiving the push.
</p>
</div>
</div>
);
}
export default function NotificationsAdminSettings() {
const form = useSettingsForm({ keys: useMemo(() => KEYS, []) });
const { data: serverChannels } = useServerNotificationChannels();
// Local draft for the relay URL; null means "show the saved value".
const [pushRelayURLDraft, setPushRelayURLDraft] = useState<string | null>(null);
if (form.isLoading) {
return (
@@ -444,10 +575,20 @@ export default function NotificationsAdminSettings() {
const webPushOn = isOn("notifications.web_push_enabled");
const emailOn = isOn("notifications.email_enabled");
const serverChannelsOn = isOn("notifications.server_channels_enabled");
// Discord and personal webhooks are opt-in (default off).
// Mobile push, Discord, and personal webhooks are opt-in (default off).
const applePushOn = form.getValue("notifications.apple_push_delivery_enabled") === "true";
const discordOn = form.getValue("notifications.discord_enabled") === "true";
const webhooksOn = form.getValue("notifications.webhooks_enabled") === "true";
// The relay URL is not part of the settings form: the server only persists
// it through the registration endpoint, alongside the credentials it mints.
const savedPushRelayURL = form.getValue("notifications.push_relay_url") || DEFAULT_PUSH_RELAY_URL;
const pushRelayURL = pushRelayURLDraft ?? savedPushRelayURL;
const pushRelayURLEdited = pushRelayURL !== savedPushRelayURL;
const pushRelayDeploymentID = form.getValue("notifications.push_relay_deployment_id");
const pushRelayAPIKeyReady = form.sensitiveConfigured.includes(
"notifications.push_relay_api_key",
);
const allowPrivate =
form.getValue("notifications.webhooks.allow_private_destinations") === "true";
// The test endpoint reads the SAVED credentials; testing with unsaved
@@ -463,7 +604,15 @@ export default function NotificationsAdminSettings() {
(key) => form.getValue(key) !== "" || form.sensitiveConfigured.includes(key),
);
const channelStates = [uiOn, webPushOn, emailOn, discordOn, webhooksOn, serverChannelsOn];
const channelStates = [
uiOn,
webPushOn,
applePushOn,
emailOn,
discordOn,
webhooksOn,
serverChannelsOn,
];
const enabledChannelCount = channelStates.filter(Boolean).length;
const failingServerChannels = (serverChannels ?? []).filter(
@@ -565,6 +714,38 @@ export default function NotificationsAdminSettings() {
onEnabledChange={setToggle("notifications.web_push_enabled")}
/>
<ChannelCard
icon={RadioTower}
title="Silo Push Relay"
description="Mobile push delivery through Silo's relay. Apple devices use APNs today; Android support will use the same relay when available."
enabled={applePushOn}
onEnabledChange={setToggle("notifications.apple_push_delivery_enabled")}
chips={
pushRelayAPIKeyReady ? (
<Chip tone="positive">Relay configured</Chip>
) : (
<Chip tone={applePushOn ? "warning" : "neutral"}>Relay registration required</Chip>
)
}
>
<div className="divide-border divide-y">
<ApplePushPrivacyDisclosure />
<SettingField
label="Relay URL"
hint="Public relay endpoint used by this Silo server; stored when you register"
type="text"
value={pushRelayURL}
onChange={(v) => setPushRelayURLDraft(v)}
/>
<RegisterRelayRow
relayURL={pushRelayURL}
deploymentID={pushRelayDeploymentID}
urlEdited={pushRelayURLEdited}
onRegistered={() => setPushRelayURLDraft(null)}
/>
</div>
</ChannelCard>
<ChannelCard
icon={Mail}
title="Email"