teg/backend/notifications.go

// 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"`
	Visible    bool            `json:"visible"`
	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 = &notificationBroker{
	clients: map[*notificationClient]struct{}{},
}

func (b *notificationBroker) subscribe(userID string) *notificationClient {
	client := &notificationClient{
		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 (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.Visible {
				switch notification.EntityType {
				case "operation":
					eventName = "operation-event"
				case "device":
					eventName = "device-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,
			visible,
			read_at,
			created_at
		FROM notifications
		WHERE user_id = $1
			AND visible = true
		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(
			&notification.ID,
			&notification.UserID,
			&notification.Type,
			&notification.Title,
			&notification.Message,
			&notification.EntityType,
			&notification.EntityID,
			&data,
			&notification.Visible,
			&readAt,
			&notification.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
		}

		var notification Notification
		var rawData []byte

		err := s.db.QueryRow(
			ctx,
			`
			INSERT INTO notifications (
				user_id,
				type,
				title,
				message,
				entity_type,
				entity_id,
				data,
				visible
			)
			VALUES (
				$1,
				$2,
				$3,
				$4,
				'operation',
				$5,
				$6::JSONB,
				true
			)
			RETURNING
				id::TEXT,
				user_id::TEXT,
				type,
				title,
				message,
				entity_type,
				COALESCE(entity_id::TEXT, ''),
				data,
				visible,
				created_at
			`,
			recipientID,
			notificationType,
			title,
			message,
			operation.ID,
			string(dataJSON),
		).Scan(
			&notification.ID,
			&notification.UserID,
			&notification.Type,
			&notification.Title,
			&notification.Message,
			&notification.EntityType,
			&notification.EntityID,
			&rawData,
			&notification.Visible,
			&notification.CreatedAt,
		)

		if err != nil {
			return err
		}

		notification.Data = json.RawMessage(rawData)

		notificationsBroker.publish(notification)
	}

	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
		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
		}

		var notification Notification
		var rawData []byte

		err := s.db.QueryRow(
			ctx,
			`
			INSERT INTO notifications (
				user_id,
				type,
				title,
				message,
				entity_type,
				entity_id,
				data,
				visible
			)
			VALUES (
				$1,
				'operation.updated',
				'Einsatz geändert',
				'',
				'operation',
				$2,
				$3::JSONB,
				false
			)
			RETURNING
				id::TEXT,
				user_id::TEXT,
				type,
				title,
				message,
				entity_type,
				COALESCE(entity_id::TEXT, ''),
				data,
				visible,
				created_at
			`,
			recipientID,
			operation.ID,
			string(dataJSON),
		).Scan(
			&notification.ID,
			&notification.UserID,
			&notification.Type,
			&notification.Title,
			&notification.Message,
			&notification.EntityType,
			&notification.EntityID,
			&rawData,
			&notification.Visible,
			&notification.CreatedAt,
		)

		if err != nil {
			return err
		}

		notification.Data = json.RawMessage(rawData)
		notificationsBroker.publish(notification)
	}

	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
		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
		}

		var notification Notification
		var rawData []byte

		err := s.db.QueryRow(
			ctx,
			`
			INSERT INTO notifications (
				user_id,
				type,
				title,
				message,
				entity_type,
				entity_id,
				data,
				visible
			)
			VALUES (
				$1,
				'operation.journal',
				'Journal aktualisiert',
				'Ein Journal-Eintrag wurde aktualisiert.',
				'operation',
				$2,
				$3::JSONB,
				false
			)
			RETURNING
				id,
				user_id::TEXT,
				type,
				title,
				message,
				entity_type,
				COALESCE(entity_id::TEXT, ''),
				data,
				visible,
				created_at
			`,
			recipientID,
			operation.ID,
			string(dataJSON),
		).Scan(
			&notification.ID,
			&notification.UserID,
			&notification.Type,
			&notification.Title,
			&notification.Message,
			&notification.EntityType,
			&notification.EntityID,
			&rawData,
			&notification.Visible,
			&notification.CreatedAt,
		)

		if err != nil {
			return err
		}

		notification.Data = json.RawMessage(rawData)
		notificationsBroker.publish(notification)
	}

	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
		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
		}

		var notification Notification
		var rawData []byte

		err := s.db.QueryRow(
			ctx,
			`
			INSERT INTO notifications (
				user_id,
				type,
				title,
				message,
				entity_type,
				entity_id,
				data,
				visible
			)
			VALUES (
				$1,
				'device.journal',
				'Geräte-Journal aktualisiert',
				'Ein Geräte-Journal-Eintrag wurde hinzugefügt.',
				'device',
				$2,
				$3::JSONB,
				false
			)
			RETURNING
				id,
				user_id::TEXT,
				type,
				title,
				message,
				entity_type,
				COALESCE(entity_id::TEXT, ''),
				data,
				visible,
				created_at
			`,
			recipientID,
			device.ID,
			string(dataJSON),
		).Scan(
			&notification.ID,
			&notification.UserID,
			&notification.Type,
			&notification.Title,
			&notification.Message,
			&notification.EntityType,
			&notification.EntityID,
			&rawData,
			&notification.Visible,
			&notification.CreatedAt,
		)

		if err != nil {
			return err
		}

		notification.Data = json.RawMessage(rawData)
		notificationsBroker.publish(notification)
	}

	return rows.Err()
}