package notifications import ( "bytes" "context" "encoding/json" "fmt" "io" "log/slog" "net/http" "os" "strings" "sync" "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 pushRelayRequestTimeout = 15 * time.Second pushRelayMaxRetryAfter = 23 * time.Hour relayAppleSendPath = "/v1/apple/send" relayFcmSendPath = "/v1/fcm/send" ) func pushRetryDelay(completedAttempt int) (time.Duration, bool) { if completedAttempt < 1 || completedAttempt >= pushMaxAttempts { return 0, false } return pushRetrySchedule[completedAttempt] - pushRetrySchedule[completedAttempt-1], true } func pushRetryDelayWithHint(completedAttempt int, retryAfter time.Duration) (time.Duration, bool) { delay, more := pushRetryDelay(completedAttempt) if retryAfter > 0 { // The relay retains idempotency state for 24 hours. Stay safely inside // that window even if APNs returns an unusually large Retry-After value. delay = min(retryAfter, pushRelayMaxRetryAfter) } return delay, more } func terminalAPNsDeviceRejection(status int, code, message string) bool { if status != http.StatusUnprocessableEntity || code != "apns_rejected" { return false } const prefix = "APNs rejected the notification:" reason := strings.TrimSpace(strings.TrimPrefix(message, prefix)) switch reason { case "BadDeviceToken", "InvalidToken", "DeviceTokenNotForTopic", "Unregistered": return true default: return false } } // terminalFCMDeviceRejection recognizes the relay's fcm_rejected responses // whose FCM error code means the registration token is permanently gone. // Request-level rejections (e.g. INVALID_ARGUMENT payload complaints) keep the // device enabled, matching the conservative APNs list above. func terminalFCMDeviceRejection(status int, code, message string) bool { if status != http.StatusUnprocessableEntity || code != "fcm_rejected" { return false } const prefix = "FCM rejected the notification:" reason := strings.TrimSpace(strings.TrimPrefix(message, prefix)) return reason == "UNREGISTERED" } func terminalDeviceRejection(platform string, status int, code, message string) bool { if platform == PushPlatformAndroid { return terminalFCMDeviceRejection(status, code, message) } return terminalAPNsDeviceRejection(status, code, message) } 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"` } // pushRelayFcmRequest matches the relay's /v1/fcm/send contract: no // environment or topic — the relay's Firebase project is the boundary. type pushRelayFcmRequest struct { Token string `json:"token"` Mode string `json:"mode"` ServerDeviceID string `json:"server_device_id"` DeliveryID string `json:"delivery_id"` CollapseID *string `json:"collapse_id,omitempty"` } 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 renewMu sync.Mutex developmentRelayURL string now func() time.Time } func newPushSender(devices *PushDeviceRepository, deliveries *DeliveryRepository, cipher *secret.Cipher, settings *Settings) *pushSender { // The Worker allows APNs up to 10 seconds. Leave enough room for edge // routing and response processing so Silo receives the relay's classified // outcome instead of manufacturing an ambiguous client-side timeout. client := newNotificationHTTPClient(nil, pushRelayRequestTimeout) return &pushSender{ devices: devices, deliveries: deliveries, cipher: cipher, settings: settings, client: client, logger: slog.Default().With("component", "notifications.apple_push"), developmentRelayURL: os.Getenv("SILO_PUSH_RELAY_DEVELOPMENT_URL"), now: time.Now, } } 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.WarnContext(ctx, "push device lookup failed", "attempt_id", attempt.ID, "error", err) } return nil } deliveryEnabled := s.settings.ApplePushDeliveryEnabled(ctx) if device.Platform == PushPlatformAndroid { deliveryEnabled = s.settings.AndroidPushDeliveryEnabled(ctx) } if !device.Enabled || device.PushMode != PushModePrivatePush || !deliveryEnabled { return s.finalize(ctx, attempt, PushOutcomeFailed, "delivery_disabled", "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.WarnContext(ctx, "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) } } ciphertext, aad := device.APNsTokenCiphertext, pushDeviceAPNsTokenAAD(device.ID) if device.Platform == PushPlatformAndroid { ciphertext, aad = device.FCMTokenCiphertext, pushDeviceFCMTokenAAD(device.ID) } token, err := s.cipher.Decrypt(ciphertext, aad) if err != nil { return s.finalize(ctx, attempt, PushOutcomeFailed, "decrypt_failed", "push 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 := pushRetryDelayWithHint(attemptNumber, 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.WarnContext(ctx, "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 { credential, err := s.prepareRelayCredential(ctx) if err != nil { return pushSendResult{HTTPStatus: http.StatusServiceUnavailable, Message: err.Error(), UpstreamReason: "relay_credential_unavailable"} } result := s.sendWithCapability(ctx, attempt, device, token, credential.RelayURL, credential.APIKey) if result.HTTPStatus != http.StatusUnauthorized || result.UpstreamReason != "token_expired" { if result.HTTPStatus == http.StatusUnauthorized { _ = s.markReregistrationRequired(ctx, credential) } return result } renewed, err := s.renewRelayCapability(ctx, credential.APIKey) if err != nil { return pushSendResult{ HTTPStatus: http.StatusServiceUnavailable, UpstreamReason: "relay_renewal_failed", Message: "push relay capability renewal failed", } } return s.sendWithCapability(ctx, attempt, device, token, renewed.RelayURL, renewed.APIKey) } func (s *pushSender) prepareRelayCredential(ctx context.Context) (PushRelayCredential, error) { s.renewMu.Lock() defer s.renewMu.Unlock() current := LoadPushRelayCredential(ctx, s.settings) relayURL, err := NormalizePushRelayURL(current.RelayURL, s.developmentRelayURL) if err != nil { return PushRelayCredential{}, err } current.RelayURL = relayURL if current.APIKey == "" { return PushRelayCredential{}, fmt.Errorf("push relay API key not configured") } if IsLegacyPushRelayKey(current.APIKey) { result, err := RegisterRelayCredential(ctx, s.settings, s.client, relayURL) return result.Credential, err } if current.ReregistrationRequired { return PushRelayCredential{}, fmt.Errorf("push relay re-registration required") } if RelayCredentialNeedsRenewal(s.now(), current.ExpiresAt, current.DeploymentID) { result, err := RenewRelayCredential(ctx, s.settings, s.client, current) // A proactive refresh must not suppress delivery while the current // capability is still valid. Reactive token_expired handling remains the // final safety net once its actual expiry is reached. if err != nil && s.now().Before(current.ExpiresAt) { return current, nil } return result.Credential, err } return current, nil } func (s *pushSender) sendWithCapability(ctx context.Context, attempt PushDeliveryAttempt, device *PushDevice, token, relayURL, apiKey string) pushSendResult { deliveryID := attempt.ID if attempt.NotificationDeliveryID != nil { deliveryID = *attempt.NotificationDeliveryID } collapseID := deliveryID sendPath := relayAppleSendPath var payload any = pushRelayAppleRequest{ Token: token, Environment: device.APNsEnvironment, Topic: device.APNsTopic, Mode: "private_alert", ServerDeviceID: device.ServerDeviceID, DeliveryID: deliveryID, CollapseID: &collapseID, } if device.Platform == PushPlatformAndroid { sendPath = relayFcmSendPath payload = pushRelayFcmRequest{ Token: token, Mode: "private_alert", ServerDeviceID: device.ServerDeviceID, DeliveryID: deliveryID, CollapseID: &collapseID, } } body, err := json.Marshal(payload) if err != nil { return pushSendResult{Message: "relay payload build failed", UpstreamReason: "payload_build_failed"} } req, err := http.NewRequestWithContext(ctx, http.MethodPost, relayURL+sendPath, 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") // One logical delivery attempt keeps the same key across every transport // retry. The relay can then distinguish a safe retry from a new delivery and // refuse to resend an ambiguous APNs outcome. req.Header.Set("Idempotency-Key", attempt.ID) 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: terminalDeviceRejection(device.Platform, resp.StatusCode, code, message), } } func (s *pushSender) renewRelayCapability(ctx context.Context, expiredKey string) (PushRelayCredential, error) { s.renewMu.Lock() defer s.renewMu.Unlock() // Another sender may have renewed while this goroutine waited for the lock. current := LoadPushRelayCredential(ctx, s.settings) if current.APIKey != "" && current.APIKey != expiredKey { return current, nil } relayURL, err := NormalizePushRelayURL(current.RelayURL, s.developmentRelayURL) if err != nil { return PushRelayCredential{}, err } current.RelayURL = relayURL result, err := RenewRelayCredential(ctx, s.settings, s.client, current) return result.Credential, err } func (s *pushSender) markReregistrationRequired(ctx context.Context, failed PushRelayCredential) error { s.renewMu.Lock() defer s.renewMu.Unlock() current := LoadPushRelayCredential(ctx, s.settings) if current.APIKey != failed.APIKey { return nil } return MarkRelayReregistrationRequired(ctx, s.settings, current) } // 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) }, // The dispatcher claims attempts for every platform; processAttempt // applies the per-platform delivery toggle to each device. enabled: sender.settings.PushDeliveryEnabled, 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) { return s.sendPushTest(ctx, PushPlatformApple, profileID, serverDeviceID) } func (s *System) SendAndroidPushTest(ctx context.Context, profileID, serverDeviceID string) (*ApplePushTestResult, error) { return s.sendPushTest(ctx, PushPlatformAndroid, profileID, serverDeviceID) } func (s *System) sendPushTest(ctx context.Context, platform, profileID, serverDeviceID string) (*ApplePushTestResult, error) { if s == nil || s.pushDeviceRepo == nil || s.pushSender == nil { return nil, ErrPushDeliveryUnavailable } deliveryEnabled := s.Settings.ApplePushDeliveryEnabled(ctx) if platform == PushPlatformAndroid { deliveryEnabled = s.Settings.AndroidPushDeliveryEnabled(ctx) } if !deliveryEnabled { return nil, ErrPushDeliveryUnavailable } if s.Settings.PushRelayAPIKey(ctx) == "" { return nil, ErrPushDeliveryUnavailable } attempt, device, err := s.pushDeviceRepo.EnqueueTestAttempt(ctx, platform, 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 }