package main import ( "context" "errors" "log" "math" "net/http" "strconv" "strings" "time" "github.com/go-chi/chi/v5" ) const ( notificationStatusQueued = "queued" notificationStatusSent = "sent" notificationStatusDeferred = "deferred" notificationStatusFailed = "failed" notificationStatusThrottled = "throttled" notificationThrottleDefer = "defer" notificationThrottleDrop = "drop" notificationThrottleDeadLetter = "dead-letter" ) type notificationRequest struct { Channel string To string Subject string Body string TemplateID string Metadata map[string]any } type notificationProviderCapabilities struct { Email bool `json:"email"` Push bool `json:"push"` Templates bool `json:"templates"` Webhooks bool `json:"webhooks"` Events bool `json:"events"` } type notificationProviderMetadata struct { Name string `json:"name"` Capabilities notificationProviderCapabilities `json:"capabilities"` } type notificationProviderResult struct { ProviderMessageID string Status string } type notificationProvider interface { Send(context.Context, notificationRequest) (notificationProviderResult, error) Metadata() notificationProviderMetadata } type noopNotificationProvider struct{} func (noopNotificationProvider) Send(ctx context.Context, request notificationRequest) (notificationProviderResult, error) { return notificationProviderResult{ ProviderMessageID: "noop-" + strings.ReplaceAll(request.To, "@", "_at_"), Status: notificationStatusQueued, }, nil } func (noopNotificationProvider) Metadata() notificationProviderMetadata { return notificationProviderMetadata{ Name: "noop", Capabilities: notificationProviderCapabilities{ Email: true, }, } } func (a *app) activeNotificationProvider() notificationProvider { if a.notificationProvider != nil { return a.notificationProvider } return noopNotificationProvider{} } func (a *app) notificationsAdminMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { claims, _ := r.Context().Value(userClaimsKey).(*accessClaims) if claims == nil { writeError(w, http.StatusUnauthorized, "no authenticated user") return } if !hasRole(claims, "admin") { writeError(w, http.StatusForbidden, "insufficient role for notification management") return } next.ServeHTTP(w, r) }) } func (a *app) listNotificationTemplates(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_templates", w, r) } func (a *app) createNotificationTemplate(w http.ResponseWriter, r *http.Request) { var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } doc, err := prepareNotificationTemplatePayload(Doc(payload), false) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_templates") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } if enforceTenant { doc["tenantId"] = tenantID } doc, err = a.create("notification_templates", doc) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create resource") return } writeJSON(w, http.StatusCreated, doc) } func (a *app) getNotificationTemplate(w http.ResponseWriter, r *http.Request) { a.getCollection("notification_templates", w, r) } func (a *app) patchNotificationTemplate(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid patch payload") return } patch, err := prepareNotificationTemplatePayload(Doc(payload), true) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_templates") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } if enforceTenant { delete(patch, "tenantId") } doc, err := a.patchWithUnsetScoped("notification_templates", id, patch, nil, tenantID, enforceTenant) if err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "resource not found") return } writeError(w, http.StatusInternalServerError, "failed to update resource") return } writeJSON(w, http.StatusOK, doc) } func (a *app) deleteNotificationTemplate(w http.ResponseWriter, r *http.Request) { a.deleteCollection("notification_templates", w, r) } func (a *app) listNotificationPreferences(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_preferences", w, r) } func (a *app) createNotificationPreference(w http.ResponseWriter, r *http.Request) { var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } doc, err := prepareNotificationPreferencePayload(Doc(payload), false) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_preferences") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } if enforceTenant { doc["tenantId"] = tenantID } doc, err = a.create("notification_preferences", doc) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create resource") return } writeJSON(w, http.StatusCreated, doc) } func (a *app) patchNotificationPreference(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid patch payload") return } patch, err := prepareNotificationPreferencePayload(Doc(payload), true) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_preferences") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } if enforceTenant { delete(patch, "tenantId") } doc, err := a.patchWithUnsetScoped("notification_preferences", id, patch, nil, tenantID, enforceTenant) if err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "resource not found") return } writeError(w, http.StatusInternalServerError, "failed to update resource") return } writeJSON(w, http.StatusOK, doc) } func (a *app) listNotificationDeliveries(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_deliveries", w, r) } func (a *app) listNotificationQueue(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_queue", w, r) } func (a *app) listNotificationDeadLetters(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_dead_letters", w, r) } func (a *app) listNotificationPolicies(w http.ResponseWriter, r *http.Request) { a.listCollectionWithRequest("notification_rate_limit_policies", w, r) } func (a *app) createNotificationPolicy(w http.ResponseWriter, r *http.Request) { tenantID, err := enforceTenantScopedMutation(r, "notification_rate_limit_policies") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } policy, err := prepareNotificationPolicyPayload(Doc(payload), false) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } policy["tenantId"] = tenantID doc, err := a.create("notification_rate_limit_policies", policy) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create policy") return } claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.policy.create", "notification_rate_limit_policies", auditResultSuccess, map[string]any{ "tenantId": tenantID, "channel": anyToString(doc["channel"]), }) writeJSON(w, http.StatusCreated, doc) } func (a *app) patchNotificationPolicy(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") tenantID, err := enforceTenantScopedMutation(r, "notification_rate_limit_policies") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } patch, err := prepareNotificationPolicyPayload(Doc(payload), true) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } delete(patch, "tenantId") doc, err := a.patchWithUnsetScoped("notification_rate_limit_policies", id, patch, nil, tenantID, true) if err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "policy not found") return } writeError(w, http.StatusInternalServerError, "failed to patch policy") return } claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.policy.patch", "notification_rate_limit_policies", auditResultSuccess, map[string]any{ "tenantId": tenantID, "policyId": id, }) writeJSON(w, http.StatusOK, doc) } func (a *app) deleteNotificationPolicy(w http.ResponseWriter, r *http.Request) { tenantID, err := enforceTenantScopedMutation(r, "notification_rate_limit_policies") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } id := chi.URLParam(r, "id") if err := a.deleteByIDScoped("notification_rate_limit_policies", id, tenantID, true); err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "policy not found") return } writeError(w, http.StatusInternalServerError, "failed to delete policy") return } claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.policy.delete", "notification_rate_limit_policies", auditResultSuccess, map[string]any{ "tenantId": tenantID, "policyId": id, }) w.WriteHeader(http.StatusNoContent) } func (a *app) listNotificationProviderConfigs(w http.ResponseWriter, r *http.Request) { tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_provider_configs") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } filter := Doc{} if enforceTenant { filter["tenantId"] = tenantID } docs, err := a.list("notification_provider_configs", filter, Sort{{Key: "updatedAt", Value: -1}}) if err != nil { writeError(w, http.StatusInternalServerError, "failed to list provider configs") return } for _, doc := range docs { delete(doc, "apiKey") } writeJSON(w, http.StatusOK, docs) } func (a *app) createNotificationProviderConfig(w http.ResponseWriter, r *http.Request) { tenantID, err := enforceTenantScopedMutation(r, "notification_provider_configs") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } doc, err := prepareNotificationProviderConfigPayload(Doc(payload), false) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } doc["tenantId"] = tenantID doc, err = a.create("notification_provider_configs", doc) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create provider config") return } delete(doc, "apiKey") claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.provider_config.create", "notification_provider_configs", auditResultSuccess, map[string]any{ "tenantId": tenantID, "provider": anyToString(doc["provider"]), }) writeJSON(w, http.StatusCreated, doc) } func (a *app) patchNotificationProviderConfig(w http.ResponseWriter, r *http.Request) { tenantID, err := enforceTenantScopedMutation(r, "notification_provider_configs") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } id := chi.URLParam(r, "id") var payload map[string]any if err := decodeJSON(r, &payload); err != nil { writeError(w, http.StatusBadRequest, "invalid payload") return } patch, err := prepareNotificationProviderConfigPayload(Doc(payload), true) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } delete(patch, "tenantId") doc, err := a.patchWithUnsetScoped("notification_provider_configs", id, patch, nil, tenantID, true) if err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "provider config not found") return } writeError(w, http.StatusInternalServerError, "failed to patch provider config") return } delete(doc, "apiKey") claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.provider_config.patch", "notification_provider_configs", auditResultSuccess, map[string]any{ "tenantId": tenantID, "id": id, }) writeJSON(w, http.StatusOK, doc) } func (a *app) getNotificationProviderHealth(w http.ResponseWriter, r *http.Request) { tenantID, _, _ := requestTenantScope(r, "notification_deliveries") active := a.activeNotificationProvider().Metadata() configured := []map[string]any{ { "name": active.Name, "capabilities": active.Capabilities, "healthy": true, "source": "platform-default", }, } if tenantID != "" { docs, err := a.list("notification_provider_configs", Doc{"tenantId": tenantID}, Sort{{Key: "updatedAt", Value: -1}}) if err == nil { for _, doc := range docs { providerName := strings.ToLower(strings.TrimSpace(anyToString(doc["provider"]))) provider := newNotificationProviderFromProfile(notificationProviderProfile{ Provider: providerName, APIKey: anyToString(doc["apiKey"]), FromEmail: anyToString(doc["fromEmail"]), APIBaseURL: anyToString(doc["apiBaseURL"]), }) meta := provider.Metadata() configured = append(configured, map[string]any{ "id": doc["id"], "name": meta.Name, "capabilities": meta.Capabilities, "healthy": true, "source": "tenant", "enabled": anyToBool(doc["enabled"]), }) } } } writeJSON(w, http.StatusOK, map[string]any{ "providers": configured, }) } func (a *app) createNotificationDelivery(w http.ResponseWriter, r *http.Request) { claims, _ := claimsFromRequest(r) var payload map[string]any if err := decodeJSON(r, &payload); err != nil { a.recordAuditEvent(r.Context(), claims, "notifications.delivery.create", "notification_deliveries", auditResultFailed, map[string]any{ "reason": "invalid_payload", }) writeError(w, http.StatusBadRequest, "invalid payload") return } req, err := prepareNotificationDeliveryRequest(Doc(payload)) if err != nil { a.recordAuditEvent(r.Context(), claims, "notifications.delivery.create", "notification_deliveries", auditResultFailed, map[string]any{ "reason": "validation_error", "message": err.Error(), }) writeError(w, http.StatusBadRequest, err.Error()) return } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_deliveries") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } if enforceTenant { if req.Metadata == nil { req.Metadata = map[string]any{} } req.Metadata["tenantId"] = tenantID } now := time.Now().UnixMilli() idempotencyKey := strings.TrimSpace(anyToString(payload["idempotencyKey"])) if idempotencyKey == "" { idempotencyKey = strings.TrimSpace(anyToString(req.Metadata["idempotencyKey"])) } delivery := Doc{ "tenantId": tenantID, "channel": req.Channel, "to": req.To, "subject": req.Subject, "body": req.Body, "templateId": req.TemplateID, "metadata": req.Metadata, "provider": strings.TrimSpace(a.cfg.NotificationProvider), "status": notificationStatusQueued, "queuedAt": now, "idempotencyKey": idempotencyKey, "compatSyncRoute": true, } delivery, err = a.create("notification_deliveries", delivery) if err != nil { a.recordAuditEvent(r.Context(), claims, "notifications.delivery.create", "notification_deliveries", auditResultFailed, map[string]any{ "channel": req.Channel, "to": req.To, "reason": "persist_failed", }) writeError(w, http.StatusInternalServerError, "failed to create delivery record") return } queueDoc := Doc{ "tenantId": tenantID, "deliveryId": anyToString(delivery["id"]), "channel": req.Channel, "to": req.To, "subject": req.Subject, "body": req.Body, "templateId": req.TemplateID, "metadata": req.Metadata, "attemptCount": 0, "nextAttemptAt": now, "status": notificationStatusQueued, "idempotencyKey": idempotencyKey, "providerHint": strings.TrimSpace(anyToString(payload["provider"])), "fallbackProvider": strings.TrimSpace(anyToString(payload["fallbackProvider"])), } queueDoc, err = a.create("notification_queue", queueDoc) if err != nil { _, _ = a.patch("notification_deliveries", anyToString(delivery["id"]), Doc{ "status": notificationStatusFailed, "lastError": "queue_insert_failed", "failedAt": time.Now().UnixMilli(), "updatedAt": time.Now().UnixMilli(), }) writeError(w, http.StatusInternalServerError, "failed to enqueue delivery") return } if shouldProcessSynchronously(payload, r) { if _, err := a.processNotificationQueueItem(r.Context(), anyToString(queueDoc["id"])); err != nil { log.Printf("sync notification queue processing failed queueId=%s err=%v", anyToString(queueDoc["id"]), err) } updated, getErr := a.getByIDScoped("notification_deliveries", anyToString(delivery["id"]), tenantID, enforceTenant) if getErr == nil { delivery = updated } } a.recordAuditEvent(r.Context(), claims, "notifications.delivery.create", "notification_deliveries", auditResultSuccess, map[string]any{ "tenantId": tenantID, "channel": req.Channel, "templateId": req.TemplateID, "to": req.To, "status": anyToString(delivery["status"]), }) writeJSON(w, http.StatusCreated, delivery) } func (a *app) processNotificationQueue(w http.ResponseWriter, r *http.Request) { limit := 20 if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" { if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 && parsed <= 100 { limit = parsed } } tenantID, enforceTenant, scopeErr := requestTenantScope(r, "notification_queue") if scopeErr != nil { writeError(w, http.StatusForbidden, scopeErr.Error()) return } col, err := a.resolveCollection("notification_queue") if err != nil { writeError(w, http.StatusInternalServerError, "queue unavailable") return } now := time.Now().UnixMilli() filter := Doc{"nextAttemptAt": Doc{"$lte": now}} if enforceTenant { filter["tenantId"] = tenantID } ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) defer cancel() cursor, err := col.Find(ctx, filter, NewFindOptions().SetSort(Sort{{Key: "nextAttemptAt", Value: 1}}).SetLimit(int64(limit))) if err != nil { writeError(w, http.StatusInternalServerError, "failed to load queue") return } defer cursor.Close(ctx) processed := 0 sent := 0 deferred := 0 failed := 0 throttled := 0 for cursor.Next(ctx) { var doc Doc if err := cursor.Decode(&doc); err != nil { continue } normalizeDocument(doc) status, err := a.processNotificationQueueItem(r.Context(), anyToString(doc["id"])) if err != nil { log.Printf("notification worker failed queueId=%s err=%v", anyToString(doc["id"]), err) continue } processed++ switch status { case notificationStatusSent: sent++ case notificationStatusDeferred: deferred++ case notificationStatusFailed: failed++ case notificationStatusThrottled: throttled++ } } writeJSON(w, http.StatusOK, map[string]any{ "processed": processed, "sent": sent, "deferred": deferred, "failed": failed, "throttled": throttled, "inspected": limit, "processedAt": time.Now().UnixMilli(), }) } func (a *app) replayNotificationDeadLetter(w http.ResponseWriter, r *http.Request) { tenantID, err := enforceTenantScopedMutation(r, "notification_dead_letters") if err != nil { writeError(w, http.StatusForbidden, err.Error()) return } id := chi.URLParam(r, "id") deadLetter, err := a.getByIDScoped("notification_dead_letters", id, tenantID, true) if err != nil { if errors.Is(err, ErrNotFound) { writeError(w, http.StatusNotFound, "dead letter not found") return } writeError(w, http.StatusInternalServerError, "failed to load dead letter") return } queueDoc := Doc{ "tenantId": tenantID, "deliveryId": anyToString(deadLetter["deliveryId"]), "channel": anyToString(deadLetter["channel"]), "to": anyToString(deadLetter["to"]), "subject": anyToString(deadLetter["subject"]), "body": anyToString(deadLetter["body"]), "templateId": anyToString(deadLetter["templateId"]), "metadata": copyAnyMap(deadLetter["metadata"]), "attemptCount": 0, "nextAttemptAt": time.Now().UnixMilli(), "status": notificationStatusQueued, "idempotencyKey": anyToString(deadLetter["idempotencyKey"]), "providerHint": anyToString(deadLetter["providerHint"]), "fallbackProvider": anyToString(deadLetter["fallbackProvider"]), "replayedFrom": id, } queueDoc, err = a.create("notification_queue", queueDoc) if err != nil { writeError(w, http.StatusInternalServerError, "failed to enqueue replay") return } if deliveryID := anyToString(deadLetter["deliveryId"]); deliveryID != "" { _, _ = a.patchWithUnsetScoped("notification_deliveries", deliveryID, Doc{ "status": notificationStatusQueued, "queuedAt": time.Now().UnixMilli(), }, nil, tenantID, true) } _, _ = a.patchWithUnsetScoped("notification_dead_letters", id, Doc{ "replayedAt": time.Now().UnixMilli(), "replayId": anyToString(queueDoc["id"]), }, nil, tenantID, true) claims, _ := claimsFromRequest(r) a.recordAuditEvent(r.Context(), claims, "notifications.dead_letter.replay", "notification_dead_letters", auditResultSuccess, map[string]any{ "tenantId": tenantID, "deadLetterId": id, "queueId": anyToString(queueDoc["id"]), }) writeJSON(w, http.StatusAccepted, queueDoc) } func (a *app) processNotificationQueueItem(ctx context.Context, queueID string) (string, error) { queueDoc, err := a.getByID("notification_queue", queueID) if err != nil { return "", err } deliveryID := anyToString(queueDoc["deliveryId"]) tenantID := strings.TrimSpace(anyToString(queueDoc["tenantId"])) delivery, err := a.getByIDScoped("notification_deliveries", deliveryID, tenantID, tenantID != "") if err == nil && anyToString(delivery["status"]) == notificationStatusSent { _ = a.deleteByID("notification_queue", queueID) return notificationStatusSent, nil } request := notificationRequest{ Channel: strings.ToLower(strings.TrimSpace(anyToString(queueDoc["channel"]))), To: strings.TrimSpace(anyToString(queueDoc["to"])), Subject: anyToString(queueDoc["subject"]), Body: anyToString(queueDoc["body"]), TemplateID: anyToString(queueDoc["templateId"]), Metadata: copyAnyMap(queueDoc["metadata"]), } policy, err := a.resolveTenantNotificationPolicy(ctx, tenantID, request.Channel) if err != nil { return "", err } throttled, throttleReason, throttleAction, err := a.isNotificationThrottled(ctx, tenantID, request, policy) if err != nil { return "", err } if throttled { status, applyErr := a.applyNotificationThrottle(queueDoc, deliveryID, throttleReason, throttleAction, policy) return status, applyErr } provider := a.resolveNotificationProvider(ctx, tenantID, anyToString(queueDoc["providerHint"]), anyToString(queueDoc["fallbackProvider"])) result, sendErr := provider.Send(ctx, request) attemptNumber := anyToInt(queueDoc["attemptCount"]) + 1 if sendErr != nil { _ = a.recordDeliveryAttempt(tenantID, deliveryID, queueID, attemptNumber, notificationStatusFailed, sendErr.Error(), "") if attemptNumber > policy.MaxRetries { if err := a.moveQueueToDeadLetter(queueDoc, deliveryID, sendErr.Error(), notificationStatusFailed); err != nil { return "", err } return notificationStatusFailed, nil } backoff := computeNotificationBackoffSeconds(policy, attemptNumber) _, _ = a.patch("notification_queue", queueID, Doc{ "attemptCount": attemptNumber, "status": notificationStatusDeferred, "lastError": sendErr.Error(), "nextAttemptAt": time.Now().Add(time.Duration(backoff) * time.Second).UnixMilli(), }) _, _ = a.patch("notification_deliveries", deliveryID, Doc{ "status": notificationStatusDeferred, "lastError": sendErr.Error(), "retryCount": attemptNumber, "provider": provider.Metadata().Name, "deferredAt": time.Now().UnixMilli(), "updatedAt": time.Now().UnixMilli(), }) return notificationStatusDeferred, nil } if result.Status == "" { result.Status = notificationStatusSent } _ = a.recordDeliveryAttempt(tenantID, deliveryID, queueID, attemptNumber, notificationStatusSent, "", result.ProviderMessageID) _, _ = a.patch("notification_deliveries", deliveryID, Doc{ "status": notificationStatusSent, "provider": provider.Metadata().Name, "providerMessageId": result.ProviderMessageID, "sentAt": time.Now().UnixMilli(), "retryCount": anyToInt(queueDoc["attemptCount"]), }) _ = a.deleteByID("notification_queue", queueID) return notificationStatusSent, nil } func (a *app) resolveNotificationProvider(ctx context.Context, tenantID, requestedProvider, fallbackProvider string) notificationProvider { primary := strings.TrimSpace(strings.ToLower(requestedProvider)) if primary == "" { primary = strings.TrimSpace(strings.ToLower(a.cfg.NotificationProvider)) } fallback := strings.TrimSpace(strings.ToLower(fallbackProvider)) if fallback == "" { fallback = "noop" } provider := a.resolveNotificationProviderProfile(ctx, tenantID, primary) if provider.Metadata().Name != "noop" { return provider } return a.resolveNotificationProviderProfile(ctx, tenantID, fallback) } func (a *app) resolveNotificationProviderProfile(ctx context.Context, tenantID, providerName string) notificationProvider { profile := notificationProviderProfile{ Provider: providerName, APIKey: a.cfg.SendGridAPIKey, FromEmail: a.cfg.SendGridFromEmail, APIBaseURL: a.cfg.SendGridAPIBaseURL, } if strings.TrimSpace(tenantID) != "" { configDoc, err := a.findActiveNotificationProviderConfig(ctx, tenantID, providerName) if err == nil && configDoc != nil { if apiKey := anyToString(configDoc["apiKey"]); apiKey != "" { profile.APIKey = apiKey } if fromEmail := anyToString(configDoc["fromEmail"]); fromEmail != "" { profile.FromEmail = fromEmail } if apiBaseURL := anyToString(configDoc["apiBaseURL"]); apiBaseURL != "" { profile.APIBaseURL = apiBaseURL } } } return newNotificationProviderFromProfile(profile) } func (a *app) findActiveNotificationProviderConfig(ctx context.Context, tenantID, providerName string) (Doc, error) { col, err := a.resolveCollection("notification_provider_configs") if err != nil { return nil, err } filter := Doc{ "tenantId": tenantID, "provider": strings.TrimSpace(strings.ToLower(providerName)), "enabled": Doc{"$ne": false}, } var doc Doc ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() if err := col.FindOne(ctxWithTimeout, filter, NewFindOneOptions().SetSort(Sort{{Key: "updatedAt", Value: -1}})).Decode(&doc); err != nil { return nil, err } normalizeDocument(doc) return doc, nil } type notificationPolicy struct { Channel string ThroughputPerMinute int Burst int MaxRetries int BackoffSeconds int MaxBackoffSeconds int ThrottledAction string DeferredSeconds int PerTemplatePerHour int PerRecipientPerHour int QuietHoursStartUTC int QuietHoursEndUTC int ChannelEnabled bool FallbackProviderName string } func (a *app) resolveTenantNotificationPolicy(ctx context.Context, tenantID, channel string) (notificationPolicy, error) { policy := defaultNotificationPolicy(channel) if strings.TrimSpace(tenantID) == "" { return policy, nil } col, err := a.resolveCollection("notification_rate_limit_policies") if err != nil { return policy, err } filter := Doc{ "tenantId": tenantID, "channel": channel, } var doc Doc ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() if err := col.FindOne(ctxWithTimeout, filter, NewFindOneOptions().SetSort(Sort{{Key: "updatedAt", Value: -1}})).Decode(&doc); err != nil { if errors.Is(err, ErrNotFound) { return policy, nil } return policy, err } normalizeDocument(doc) policy.ThroughputPerMinute = defaultIfZero(anyToInt(doc["throughputPerMinute"]), policy.ThroughputPerMinute) policy.Burst = defaultIfZero(anyToInt(doc["burst"]), policy.Burst) policy.MaxRetries = defaultIfNegative(anyToInt(doc["maxRetries"]), policy.MaxRetries) policy.BackoffSeconds = defaultIfZero(anyToInt(doc["backoffSeconds"]), policy.BackoffSeconds) policy.MaxBackoffSeconds = defaultIfZero(anyToInt(doc["maxBackoffSeconds"]), policy.MaxBackoffSeconds) policy.DeferredSeconds = defaultIfZero(anyToInt(doc["deferredSeconds"]), policy.DeferredSeconds) policy.PerTemplatePerHour = defaultIfNegative(anyToInt(doc["perTemplatePerHour"]), policy.PerTemplatePerHour) policy.PerRecipientPerHour = defaultIfNegative(anyToInt(doc["perRecipientPerHour"]), policy.PerRecipientPerHour) policy.QuietHoursStartUTC = normalizeHour(anyToInt(doc["quietHoursStartUtc"])) policy.QuietHoursEndUTC = normalizeHour(anyToInt(doc["quietHoursEndUtc"])) policy.ChannelEnabled = true if _, has := doc["channelEnabled"]; has { policy.ChannelEnabled = anyToBool(doc["channelEnabled"]) } if throttle := strings.TrimSpace(strings.ToLower(anyToString(doc["throttledAction"]))); throttle != "" { policy.ThrottledAction = throttle } if fallback := strings.TrimSpace(strings.ToLower(anyToString(doc["fallbackProvider"]))); fallback != "" { policy.FallbackProviderName = fallback } return policy, nil } func defaultNotificationPolicy(channel string) notificationPolicy { return notificationPolicy{ Channel: strings.TrimSpace(strings.ToLower(channel)), ThroughputPerMinute: 60, Burst: 20, MaxRetries: 3, BackoffSeconds: 30, MaxBackoffSeconds: 600, ThrottledAction: notificationThrottleDefer, DeferredSeconds: 60, PerTemplatePerHour: 240, PerRecipientPerHour: 60, QuietHoursStartUTC: -1, QuietHoursEndUTC: -1, ChannelEnabled: true, FallbackProviderName: "noop", } } func (a *app) isNotificationThrottled(ctx context.Context, tenantID string, request notificationRequest, policy notificationPolicy) (bool, string, string, error) { if !policy.ChannelEnabled { return true, "channel_disabled", notificationThrottleDrop, nil } if inQuietHours(policy, time.Now().UTC()) { return true, "quiet_hours", notificationThrottleDefer, nil } if tenantID == "" { return false, "", "", nil } now := time.Now() if policy.ThroughputPerMinute > 0 { count, err := a.countCollection(ctx, "notification_delivery_attempts", Doc{ "tenantId": tenantID, "channel": request.Channel, "createdAt": Doc{"$gte": now.Add(-1 * time.Minute).UnixMilli()}, "attemptStatus": Doc{"$in": []string{notificationStatusSent, notificationStatusFailed}}, }) if err != nil { return false, "", "", err } if count >= int64(policy.ThroughputPerMinute+policy.Burst) { return true, "tenant_channel_throughput", policy.ThrottledAction, nil } } if policy.PerTemplatePerHour > 0 && request.TemplateID != "" { count, err := a.countCollection(ctx, "notification_deliveries", Doc{ "tenantId": tenantID, "templateId": request.TemplateID, "queuedAt": Doc{"$gte": now.Add(-1 * time.Hour).UnixMilli()}, }) if err != nil { return false, "", "", err } if count >= int64(policy.PerTemplatePerHour) { return true, "template_safeguard", policy.ThrottledAction, nil } } if policy.PerRecipientPerHour > 0 && request.To != "" { count, err := a.countCollection(ctx, "notification_deliveries", Doc{ "tenantId": tenantID, "to": request.To, "queuedAt": Doc{"$gte": now.Add(-1 * time.Hour).UnixMilli()}, }) if err != nil { return false, "", "", err } if count >= int64(policy.PerRecipientPerHour) { return true, "recipient_safeguard", policy.ThrottledAction, nil } } return false, "", "", nil } func (a *app) applyNotificationThrottle(queueDoc Doc, deliveryID, reason, action string, policy notificationPolicy) (string, error) { queueID := anyToString(queueDoc["id"]) now := time.Now().UnixMilli() normalizedAction := strings.TrimSpace(strings.ToLower(action)) if normalizedAction == "" { normalizedAction = notificationThrottleDefer } switch normalizedAction { case notificationThrottleDrop: _, _ = a.patch("notification_deliveries", deliveryID, Doc{ "status": notificationStatusThrottled, "throttleReason": reason, "throttledAt": now, }) _ = a.deleteByID("notification_queue", queueID) return notificationStatusThrottled, nil case notificationThrottleDeadLetter: if err := a.moveQueueToDeadLetter(queueDoc, deliveryID, reason, notificationStatusThrottled); err != nil { return "", err } return notificationStatusThrottled, nil default: nextAttemptAt := time.Now().Add(time.Duration(policy.DeferredSeconds) * time.Second).UnixMilli() _, _ = a.patch("notification_queue", queueID, Doc{ "status": notificationStatusDeferred, "throttleReason": reason, "nextAttemptAt": nextAttemptAt, }) _, _ = a.patch("notification_deliveries", deliveryID, Doc{ "status": notificationStatusDeferred, "throttleReason": reason, "deferredAt": now, }) return notificationStatusDeferred, nil } } func (a *app) moveQueueToDeadLetter(queueDoc Doc, deliveryID, reason, status string) error { now := time.Now().UnixMilli() deadLetter := Doc{ "tenantId": anyToString(queueDoc["tenantId"]), "deliveryId": deliveryID, "queueId": anyToString(queueDoc["id"]), "channel": anyToString(queueDoc["channel"]), "to": anyToString(queueDoc["to"]), "subject": anyToString(queueDoc["subject"]), "body": anyToString(queueDoc["body"]), "templateId": anyToString(queueDoc["templateId"]), "metadata": copyAnyMap(queueDoc["metadata"]), "idempotencyKey": anyToString(queueDoc["idempotencyKey"]), "providerHint": anyToString(queueDoc["providerHint"]), "fallbackProvider": anyToString(queueDoc["fallbackProvider"]), "attemptCount": anyToInt(queueDoc["attemptCount"]), "reason": reason, "movedAt": now, } if _, err := a.create("notification_dead_letters", deadLetter); err != nil { return err } if deliveryID != "" { _, _ = a.patch("notification_deliveries", deliveryID, Doc{ "status": status, "lastError": reason, "failedAt": now, "throttledAt": now, }) } return a.deleteByID("notification_queue", anyToString(queueDoc["id"])) } func (a *app) recordDeliveryAttempt(tenantID, deliveryID, queueID string, attempt int, attemptStatus, errorMessage, providerMessageID string) error { _, err := a.create("notification_delivery_attempts", Doc{ "tenantId": tenantID, "deliveryId": deliveryID, "queueId": queueID, "attempt": attempt, "attemptStatus": attemptStatus, "errorMessage": errorMessage, "providerMessageId": providerMessageID, "createdAt": time.Now().UnixMilli(), }) return err } func (a *app) countCollection(ctx context.Context, name string, filter Doc) (int64, error) { col, err := a.resolveCollection(name) if err != nil { return 0, err } ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() return col.CountDocuments(ctxWithTimeout, filter) } func enforceTenantScopedMutation(r *http.Request, collection string) (string, error) { tenantID, enforceTenant, err := requestTenantScope(r, collection) if err != nil { return "", err } if !enforceTenant || strings.TrimSpace(tenantID) == "" { return "", errors.New("tenant override required for mutation") } return tenantID, nil } func shouldProcessSynchronously(payload map[string]any, r *http.Request) bool { if rawMode := strings.TrimSpace(strings.ToLower(r.URL.Query().Get("mode"))); rawMode != "" { return rawMode != "async" } if asyncVal, has := payload["async"]; has && anyToBool(asyncVal) { return false } if syncVal, has := payload["sync"]; has { return anyToBool(syncVal) } return true } func computeNotificationBackoffSeconds(policy notificationPolicy, attempt int) int { if attempt < 1 { attempt = 1 } base := policy.BackoffSeconds if base <= 0 { base = 1 } backoff := int(float64(base) * math.Pow(2, float64(attempt-1))) if policy.MaxBackoffSeconds > 0 && backoff > policy.MaxBackoffSeconds { return policy.MaxBackoffSeconds } return backoff } func inQuietHours(policy notificationPolicy, now time.Time) bool { start := normalizeHour(policy.QuietHoursStartUTC) end := normalizeHour(policy.QuietHoursEndUTC) if start < 0 || end < 0 || start == end { return false } hour := now.Hour() if start < end { return hour >= start && hour < end } return hour >= start || hour < end } func normalizeHour(value int) int { if value < 0 || value > 23 { return -1 } return value } func copyAnyMap(raw any) map[string]any { typed, ok := raw.(map[string]any) if !ok { if docTyped, docOK := raw.(Doc); docOK { typed = map[string]any(docTyped) } else { return map[string]any{} } } out := make(map[string]any, len(typed)) for key, value := range typed { out[key] = value } return out } func anyToBool(value any) bool { switch typed := value.(type) { case bool: return typed case string: trimmed := strings.TrimSpace(strings.ToLower(typed)) return trimmed == "1" || trimmed == "true" || trimmed == "yes" || trimmed == "on" case float64: return typed != 0 case int: return typed != 0 default: return false } } func anyToInt(value any) int { switch typed := value.(type) { case int: return typed case int32: return int(typed) case int64: return int(typed) case float64: return int(typed) case string: parsed, err := strconv.Atoi(strings.TrimSpace(typed)) if err != nil { return 0 } return parsed default: return 0 } } func defaultIfZero(value, fallback int) int { if value == 0 { return fallback } return value } func defaultIfNegative(value, fallback int) int { if value < 0 { return fallback } return value } func prepareNotificationTemplatePayload(payload Doc, partial bool) (Doc, error) { name, hasName, err := getOptionalString(payload, "name") if err != nil { return nil, errors.New("name must be a string") } if !partial && (!hasName || name == "") { return nil, errors.New("name is required") } if hasName { payload["name"] = name } channel, hasChannel, err := getOptionalString(payload, "channel") if err != nil { return nil, errors.New("channel must be a string") } if !partial && !hasChannel { channel = "email" hasChannel = true } if hasChannel { channel = strings.ToLower(strings.TrimSpace(channel)) if channel == "" { channel = "email" } payload["channel"] = channel } for _, field := range []string{"subject", "body"} { value, has, err := getOptionalString(payload, field) if err != nil { return nil, errors.New(field + " must be a string") } if has { payload[field] = value } } enabledRaw, hasEnabled := payload["enabled"] if !hasEnabled && !partial { payload["enabled"] = true } else if hasEnabled { enabled, ok := enabledRaw.(bool) if !ok { return nil, errors.New("enabled must be a boolean") } payload["enabled"] = enabled } return payload, nil } func prepareNotificationPreferencePayload(payload Doc, partial bool) (Doc, error) { userID, hasUserID, err := getOptionalString(payload, "userId") if err != nil { return nil, errors.New("userId must be a string") } if !partial && (!hasUserID || userID == "") { return nil, errors.New("userId is required") } if hasUserID { payload["userId"] = userID } for _, field := range []string{"email", "inApp", "sms", "push"} { value, has := payload[field] if !has { if !partial { payload[field] = field != "sms" } continue } boolean, ok := value.(bool) if !ok { return nil, errors.New(field + " must be a boolean") } payload[field] = boolean } return payload, nil } func prepareNotificationPolicyPayload(payload Doc, partial bool) (Doc, error) { channel, hasChannel, err := getOptionalString(payload, "channel") if err != nil { return nil, errors.New("channel must be a string") } if !partial && strings.TrimSpace(channel) == "" && !hasChannel { channel = "email" hasChannel = true } if hasChannel { channel = strings.TrimSpace(strings.ToLower(channel)) if channel == "" { channel = "email" } payload["channel"] = channel } for _, field := range []string{ "throughputPerMinute", "burst", "maxRetries", "backoffSeconds", "maxBackoffSeconds", "deferredSeconds", "perTemplatePerHour", "perRecipientPerHour", "quietHoursStartUtc", "quietHoursEndUtc", } { raw, has := payload[field] if !has { continue } payload[field] = anyToInt(raw) } if raw, has := payload["channelEnabled"]; has { payload["channelEnabled"] = anyToBool(raw) } if raw, has := payload["throttledAction"]; has { action := strings.TrimSpace(strings.ToLower(anyToString(raw))) switch action { case notificationThrottleDefer, notificationThrottleDrop, notificationThrottleDeadLetter: payload["throttledAction"] = action default: return nil, errors.New("throttledAction must be one of defer, drop, dead-letter") } } if raw, has := payload["fallbackProvider"]; has { payload["fallbackProvider"] = strings.TrimSpace(strings.ToLower(anyToString(raw))) } return payload, nil } func prepareNotificationProviderConfigPayload(payload Doc, partial bool) (Doc, error) { provider, hasProvider, err := getOptionalString(payload, "provider") if err != nil { return nil, errors.New("provider must be a string") } if !partial && (!hasProvider || provider == "") { return nil, errors.New("provider is required") } if hasProvider { payload["provider"] = strings.TrimSpace(strings.ToLower(provider)) } for _, field := range []string{"apiKey", "fromEmail", "apiBaseURL", "credentialMode"} { value, has, err := getOptionalString(payload, field) if err != nil { return nil, errors.New(field + " must be a string") } if has { payload[field] = value } } if raw, has := payload["enabled"]; has { payload["enabled"] = anyToBool(raw) } else if !partial { payload["enabled"] = true } return payload, nil } func prepareNotificationDeliveryRequest(payload Doc) (notificationRequest, error) { to, hasTo, err := getOptionalString(payload, "to") if err != nil { return notificationRequest{}, errors.New("to must be a string") } if !hasTo || to == "" { return notificationRequest{}, errors.New("to is required") } channel, _, err := getOptionalString(payload, "channel") if err != nil { return notificationRequest{}, errors.New("channel must be a string") } if channel == "" { channel = "email" } subject, _, err := getOptionalString(payload, "subject") if err != nil { return notificationRequest{}, errors.New("subject must be a string") } body, _, err := getOptionalString(payload, "body") if err != nil { return notificationRequest{}, errors.New("body must be a string") } templateID, _, err := getOptionalString(payload, "templateId") if err != nil { return notificationRequest{}, errors.New("templateId must be a string") } if subject == "" && body == "" && templateID == "" { return notificationRequest{}, errors.New("subject/body or templateId is required") } metadata := map[string]any{} if metadataRaw, has := payload["metadata"]; has { typed, ok := metadataRaw.(map[string]any) if !ok { if docTyped, docOK := metadataRaw.(Doc); docOK { typed = map[string]any(docTyped) } else { return notificationRequest{}, errors.New("metadata must be an object") } } metadata = typed } return notificationRequest{ Channel: strings.ToLower(strings.TrimSpace(channel)), To: to, Subject: subject, Body: body, TemplateID: templateID, Metadata: metadata, }, nil }