// backend/notifications.go package main import ( "context" "database/sql" "encoding/json" "fmt" "net/http" "strings" "sync" "time" ) type Notification struct { ID string `json:"id"` UserID string `json:"userId"` Type string `json:"type"` Title string `json:"title"` Message string `json:"message"` EntityType string `json:"entityType"` EntityID string `json:"entityId"` Data json.RawMessage `json:"data"` Silent bool `json:"silent"` SoundName string `json:"soundName,omitempty"` ReadAt *time.Time `json:"readAt"` CreatedAt time.Time `json:"createdAt"` } type notificationClient struct { userID string ch chan Notification } type notificationBroker struct { mu sync.RWMutex clients map[*notificationClient]struct{} } var notificationsBroker = ¬ificationBroker{ clients: map[*notificationClient]struct{}{}, } func (b *notificationBroker) subscribe(userID string) *notificationClient { client := ¬ificationClient{ userID: userID, ch: make(chan Notification, 16), } b.mu.Lock() b.clients[client] = struct{}{} b.mu.Unlock() return client } func (b *notificationBroker) unsubscribe(client *notificationClient) { b.mu.Lock() delete(b.clients, client) close(client.ch) b.mu.Unlock() } func (b *notificationBroker) publish(notification Notification) { b.mu.RLock() defer b.mu.RUnlock() for client := range b.clients { if client.userID != notification.UserID { continue } select { case client.ch <- notification: default: } } } func (b *notificationBroker) publishToAll(notification Notification) { b.mu.RLock() defer b.mu.RUnlock() for client := range b.clients { select { case client.ch <- notification: default: } } } func shouldDeliverNotification( preferences notificationPreferencesResponse, notificationType string, entityType string, ) bool { if !preferences.InAppEnabled { return false } switch { case notificationType == "operation.journal": return preferences.OperationJournals case strings.HasPrefix(notificationType, "operation.") || entityType == "operation": return preferences.OperationUpdates case notificationType == "device.journal": return preferences.DeviceJournals case strings.HasPrefix(notificationType, "device.") || entityType == "device": return preferences.DeviceUpdates case strings.HasPrefix(notificationType, "system.") || entityType == "system": return preferences.SystemMessages default: return true } } func notificationSoundName(preferences notificationPreferencesResponse) string { if !preferences.SoundEnabled { return "" } return normalizeNotificationSoundName(preferences.SoundName) } func (s *Server) createAndPublishNotification( ctx context.Context, recipientID string, notificationType string, title string, message string, entityType string, entityID string, dataJSON []byte, silent bool, ) error { preferences, err := s.getNotificationPreferencesOrDefault(ctx, recipientID) if err != nil { return err } // Sichtbare Benachrichtigungen respektieren die User-Preferences. // Stille Events werden immer erzeugt, weil sie das UI synchronisieren. if !silent && !shouldDeliverNotification(preferences, notificationType, entityType) { return nil } var notification Notification var rawData []byte err = s.db.QueryRow( ctx, ` INSERT INTO notifications ( user_id, type, title, message, entity_type, entity_id, data, silent ) VALUES ( $1, $2, $3, $4, $5, NULLIF($6, '')::UUID, $7::JSONB, $8 ) RETURNING id::TEXT, user_id::TEXT, type, title, message, entity_type, COALESCE(entity_id::TEXT, ''), data, silent, created_at `, recipientID, notificationType, title, message, entityType, entityID, string(dataJSON), silent, ).Scan( ¬ification.ID, ¬ification.UserID, ¬ification.Type, ¬ification.Title, ¬ification.Message, ¬ification.EntityType, ¬ification.EntityID, &rawData, ¬ification.Silent, ¬ification.CreatedAt, ) if err != nil { return err } notification.Data = json.RawMessage(rawData) if !notification.Silent { notification.SoundName = notificationSoundName(preferences) } notificationsBroker.publish(notification) return nil } func (s *Server) handleNotificationEvents(w http.ResponseWriter, r *http.Request) { user, ok := userFromContext(r.Context()) if !ok { writeError(w, http.StatusUnauthorized, "Unauthorized") return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") flusher, ok := w.(http.Flusher) if !ok { writeError(w, http.StatusInternalServerError, "Streaming wird nicht unterstützt") return } client := notificationsBroker.subscribe(user.ID) defer notificationsBroker.unsubscribe(client) fmt.Fprintf(w, "event: connected\n") fmt.Fprintf(w, "data: {}\n\n") flusher.Flush() for { select { case <-r.Context().Done(): return case notification := <-client.ch: payload, err := json.Marshal(notification) if err != nil { continue } eventName := "notification" if notification.Silent { switch notification.EntityType { case "operation": eventName = "operation-event" case "device": eventName = "device-event" case "user": eventName = "user-event" default: eventName = "notification-event" } } fmt.Fprintf(w, "event: %s\n", eventName) fmt.Fprintf(w, "data: %s\n\n", payload) flusher.Flush() } } } func (s *Server) handleListNotifications(w http.ResponseWriter, r *http.Request) { user, ok := userFromContext(r.Context()) if !ok { writeError(w, http.StatusUnauthorized, "Unauthorized") return } rows, err := s.db.Query( r.Context(), ` SELECT id::TEXT, user_id::TEXT, type, title, message, entity_type, COALESCE(entity_id::TEXT, ''), data, silent, read_at, created_at FROM notifications WHERE user_id = $1 AND silent = false ORDER BY created_at DESC LIMIT 50 `, user.ID, ) if err != nil { writeError(w, http.StatusInternalServerError, "Benachrichtigungen konnten nicht geladen werden") return } defer rows.Close() notifications := []Notification{} for rows.Next() { var notification Notification var readAt sql.NullTime var data []byte if err := rows.Scan( ¬ification.ID, ¬ification.UserID, ¬ification.Type, ¬ification.Title, ¬ification.Message, ¬ification.EntityType, ¬ification.EntityID, &data, ¬ification.Silent, &readAt, ¬ification.CreatedAt, ); err != nil { writeError(w, http.StatusInternalServerError, "Benachrichtigungen konnten nicht gelesen werden") return } if readAt.Valid { notification.ReadAt = &readAt.Time } notification.Data = json.RawMessage(data) notifications = append(notifications, notification) } writeJSON(w, http.StatusOK, map[string]any{ "notifications": notifications, }) } func (s *Server) handleMarkNotificationRead(w http.ResponseWriter, r *http.Request) { user, ok := userFromContext(r.Context()) if !ok { writeError(w, http.StatusUnauthorized, "Unauthorized") return } notificationID := strings.TrimSpace(r.PathValue("id")) if notificationID == "" { writeError(w, http.StatusBadRequest, "Benachrichtigungs-ID fehlt") return } _, err := s.db.Exec( r.Context(), ` UPDATE notifications SET read_at = COALESCE(read_at, now()) WHERE id = $1 AND user_id = $2 `, notificationID, user.ID, ) if err != nil { writeError(w, http.StatusInternalServerError, "Benachrichtigung konnte nicht aktualisiert werden") return } writeJSON(w, http.StatusOK, map[string]any{ "ok": true, }) } func (s *Server) handleMarkAllNotificationsRead(w http.ResponseWriter, r *http.Request) { user, ok := userFromContext(r.Context()) if !ok { writeError(w, http.StatusUnauthorized, "Unauthorized") return } _, err := s.db.Exec( r.Context(), ` UPDATE notifications SET read_at = COALESCE(read_at, now()) WHERE user_id = $1 AND read_at IS NULL `, user.ID, ) if err != nil { writeError(w, http.StatusInternalServerError, "Benachrichtigungen konnten nicht aktualisiert werden") return } writeJSON(w, http.StatusOK, map[string]any{ "ok": true, }) } func (s *Server) createOperationNotification( ctx context.Context, actor User, operation Operation, notificationType string, title string, message string, data map[string]any, ) error { dataJSON, err := json.Marshal(data) if err != nil { return err } rows, err := s.db.Query( ctx, ` SELECT id::TEXT FROM users WHERE id <> $1 AND ( 'admin' = ANY(rights) OR 'operations:read' = ANY(rights) ) `, actor.ID, ) if err != nil { return err } defer rows.Close() for rows.Next() { var recipientID string if err := rows.Scan(&recipientID); err != nil { return err } if err := s.createAndPublishNotification( ctx, recipientID, notificationType, title, message, "operation", operation.ID, dataJSON, false, ); err != nil { return err } } return rows.Err() } func (s *Server) createOperationCreatedEvent( ctx context.Context, actor User, operation Operation, data map[string]any, ) error { dataJSON, err := json.Marshal(data) if err != nil { return err } rows, err := s.db.Query( ctx, ` SELECT id::TEXT FROM users WHERE id <> $1 AND ( 'admin' = ANY(rights) OR 'operations:read' = ANY(rights) ) `, actor.ID, ) if err != nil { return err } defer rows.Close() for rows.Next() { var recipientID string if err := rows.Scan(&recipientID); err != nil { return err } if err := s.createAndPublishNotification( ctx, recipientID, "operation.created", "Einsatz erstellt", "", "operation", operation.ID, dataJSON, true, ); err != nil { return err } } return rows.Err() } func (s *Server) createOperationUpdateEvent( ctx context.Context, actor User, operation Operation, data map[string]any, ) error { dataJSON, err := json.Marshal(data) if err != nil { return err } rows, err := s.db.Query( ctx, ` SELECT id::TEXT FROM users WHERE id <> $1 AND ( 'admin' = ANY(rights) OR 'operations:read' = ANY(rights) ) `, actor.ID, ) if err != nil { return err } defer rows.Close() for rows.Next() { var recipientID string if err := rows.Scan(&recipientID); err != nil { return err } if err := s.createAndPublishNotification( ctx, recipientID, "operation.updated", "Einsatz geändert", "", "operation", operation.ID, dataJSON, true, ); err != nil { return err } } return rows.Err() } func (s *Server) createOperationJournalEvent( ctx context.Context, actor User, operation Operation, data map[string]any, ) error { dataJSON, err := json.Marshal(data) if err != nil { return err } rows, err := s.db.Query( ctx, ` SELECT id::TEXT FROM users WHERE id <> $1 `, actor.ID, ) if err != nil { return err } defer rows.Close() for rows.Next() { var recipientID string if err := rows.Scan(&recipientID); err != nil { return err } if err := s.createAndPublishNotification( ctx, recipientID, "operation.journal", "Journal aktualisiert", "Ein Journal-Eintrag wurde aktualisiert.", "operation", operation.ID, dataJSON, true, ); err != nil { return err } } return rows.Err() } func (s *Server) createDeviceJournalEvent( ctx context.Context, actor User, device Device, data map[string]any, ) error { dataJSON, err := json.Marshal(data) if err != nil { return err } rows, err := s.db.Query( ctx, ` SELECT id::TEXT FROM users WHERE id <> $1 AND ( 'admin' = ANY(rights) OR 'devices:read' = ANY(rights) ) `, actor.ID, ) if err != nil { return err } defer rows.Close() for rows.Next() { var recipientID string if err := rows.Scan(&recipientID); err != nil { return err } if err := s.createAndPublishNotification( ctx, recipientID, "device.journal", "Geräte-Journal aktualisiert", "Ein Geräte-Journal-Eintrag wurde hinzugefügt.", "device", device.ID, dataJSON, true, ); err != nil { return err } } return rows.Err() }