Files
breakpilot-core/consent-service/internal/services/notification_service.go
Benjamin Boenisch ad111d5e69 Initial commit: breakpilot-core - Shared Infrastructure
Docker Compose with 24+ services:
- PostgreSQL (PostGIS), Valkey, MinIO, Qdrant
- Vault (PKI/TLS), Nginx (Reverse Proxy)
- Backend Core API, Consent Service, Billing Service
- RAG Service, Embedding Service
- Gitea, Woodpecker CI/CD
- Night Scheduler, Health Aggregator
- Jitsi (Web/XMPP/JVB/Jicofo), Mailpit

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-11 23:47:13 +01:00

348 lines
12 KiB
Go

package services
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
)
// NotificationType defines the type of notification
type NotificationType string
const (
NotificationTypeConsentRequired NotificationType = "consent_required"
NotificationTypeConsentReminder NotificationType = "consent_reminder"
NotificationTypeVersionPublished NotificationType = "version_published"
NotificationTypeVersionApproved NotificationType = "version_approved"
NotificationTypeVersionRejected NotificationType = "version_rejected"
NotificationTypeAccountSuspended NotificationType = "account_suspended"
NotificationTypeAccountRestored NotificationType = "account_restored"
NotificationTypeGeneral NotificationType = "general"
// DSR (Data Subject Request) notification types
NotificationTypeDSRReceived NotificationType = "dsr_received"
NotificationTypeDSRAssigned NotificationType = "dsr_assigned"
NotificationTypeDSRDeadline NotificationType = "dsr_deadline"
)
// NotificationChannel defines how notification is delivered
type NotificationChannel string
const (
ChannelInApp NotificationChannel = "in_app"
ChannelEmail NotificationChannel = "email"
ChannelPush NotificationChannel = "push"
)
// Notification represents a notification entity
type Notification struct {
ID uuid.UUID `json:"id"`
UserID uuid.UUID `json:"user_id"`
Type NotificationType `json:"type"`
Channel NotificationChannel `json:"channel"`
Title string `json:"title"`
Body string `json:"body"`
Data map[string]interface{} `json:"data,omitempty"`
ReadAt *time.Time `json:"read_at,omitempty"`
SentAt *time.Time `json:"sent_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
// NotificationPreferences holds user notification settings
type NotificationPreferences struct {
UserID uuid.UUID `json:"user_id"`
EmailEnabled bool `json:"email_enabled"`
PushEnabled bool `json:"push_enabled"`
InAppEnabled bool `json:"in_app_enabled"`
ReminderFrequency string `json:"reminder_frequency"`
}
// NotificationService handles notification operations
type NotificationService struct {
pool *pgxpool.Pool
emailService *EmailService
}
// NewNotificationService creates a new notification service
func NewNotificationService(pool *pgxpool.Pool, emailService *EmailService) *NotificationService {
return &NotificationService{
pool: pool,
emailService: emailService,
}
}
// CreateNotification creates and optionally sends a notification
func (s *NotificationService) CreateNotification(ctx context.Context, userID uuid.UUID, notifType NotificationType, title, body string, data map[string]interface{}) error {
// Get user preferences
prefs, err := s.GetPreferences(ctx, userID)
if err != nil {
// Use default preferences if not found
prefs = &NotificationPreferences{
UserID: userID,
EmailEnabled: true,
PushEnabled: true,
InAppEnabled: true,
}
}
// Create in-app notification if enabled
if prefs.InAppEnabled {
if err := s.createInAppNotification(ctx, userID, notifType, title, body, data); err != nil {
return fmt.Errorf("failed to create in-app notification: %w", err)
}
}
// Send email notification if enabled
if prefs.EmailEnabled && s.emailService != nil {
go s.sendEmailNotification(ctx, userID, notifType, title, body, data)
}
// Push notification would be sent here if enabled
// if prefs.PushEnabled {
// go s.sendPushNotification(ctx, userID, title, body, data)
// }
return nil
}
// createInAppNotification creates an in-app notification
func (s *NotificationService) createInAppNotification(ctx context.Context, userID uuid.UUID, notifType NotificationType, title, body string, data map[string]interface{}) error {
dataJSON, _ := json.Marshal(data)
_, err := s.pool.Exec(ctx, `
INSERT INTO notifications (user_id, type, channel, title, body, data, created_at)
VALUES ($1, $2, $3, $4, $5, $6, NOW())
`, userID, notifType, ChannelInApp, title, body, dataJSON)
return err
}
// sendEmailNotification sends an email notification
func (s *NotificationService) sendEmailNotification(ctx context.Context, userID uuid.UUID, notifType NotificationType, title, body string, data map[string]interface{}) {
// Get user email
var email string
err := s.pool.QueryRow(ctx, `SELECT email FROM users WHERE id = $1`, userID).Scan(&email)
if err != nil {
return
}
// Send based on notification type
switch notifType {
case NotificationTypeConsentRequired, NotificationTypeConsentReminder:
s.emailService.SendConsentReminderEmail(email, title, body)
default:
s.emailService.SendGenericNotificationEmail(email, title, body)
}
// Mark as sent
s.pool.Exec(ctx, `
UPDATE notifications SET sent_at = NOW()
WHERE user_id = $1 AND type = $2 AND channel = $3 AND sent_at IS NULL
ORDER BY created_at DESC LIMIT 1
`, userID, notifType, ChannelEmail)
}
// GetUserNotifications returns notifications for a user
func (s *NotificationService) GetUserNotifications(ctx context.Context, userID uuid.UUID, limit, offset int, unreadOnly bool) ([]Notification, int, error) {
// Count total
var totalQuery string
var total int
if unreadOnly {
totalQuery = `SELECT COUNT(*) FROM notifications WHERE user_id = $1 AND read_at IS NULL`
} else {
totalQuery = `SELECT COUNT(*) FROM notifications WHERE user_id = $1`
}
s.pool.QueryRow(ctx, totalQuery, userID).Scan(&total)
// Get notifications
var query string
if unreadOnly {
query = `
SELECT id, user_id, type, channel, title, body, data, read_at, sent_at, created_at
FROM notifications
WHERE user_id = $1 AND read_at IS NULL
ORDER BY created_at DESC
LIMIT $2 OFFSET $3
`
} else {
query = `
SELECT id, user_id, type, channel, title, body, data, read_at, sent_at, created_at
FROM notifications
WHERE user_id = $1
ORDER BY created_at DESC
LIMIT $2 OFFSET $3
`
}
rows, err := s.pool.Query(ctx, query, userID, limit, offset)
if err != nil {
return nil, 0, err
}
defer rows.Close()
var notifications []Notification
for rows.Next() {
var n Notification
var dataJSON []byte
if err := rows.Scan(&n.ID, &n.UserID, &n.Type, &n.Channel, &n.Title, &n.Body, &dataJSON, &n.ReadAt, &n.SentAt, &n.CreatedAt); err != nil {
continue
}
if dataJSON != nil {
json.Unmarshal(dataJSON, &n.Data)
}
notifications = append(notifications, n)
}
return notifications, total, nil
}
// GetUnreadCount returns the count of unread notifications
func (s *NotificationService) GetUnreadCount(ctx context.Context, userID uuid.UUID) (int, error) {
var count int
err := s.pool.QueryRow(ctx, `
SELECT COUNT(*) FROM notifications WHERE user_id = $1 AND read_at IS NULL
`, userID).Scan(&count)
return count, err
}
// MarkAsRead marks a notification as read
func (s *NotificationService) MarkAsRead(ctx context.Context, userID uuid.UUID, notificationID uuid.UUID) error {
result, err := s.pool.Exec(ctx, `
UPDATE notifications SET read_at = NOW()
WHERE id = $1 AND user_id = $2 AND read_at IS NULL
`, notificationID, userID)
if err != nil {
return err
}
if result.RowsAffected() == 0 {
return fmt.Errorf("notification not found or already read")
}
return nil
}
// MarkAllAsRead marks all notifications as read for a user
func (s *NotificationService) MarkAllAsRead(ctx context.Context, userID uuid.UUID) error {
_, err := s.pool.Exec(ctx, `
UPDATE notifications SET read_at = NOW()
WHERE user_id = $1 AND read_at IS NULL
`, userID)
return err
}
// DeleteNotification deletes a notification
func (s *NotificationService) DeleteNotification(ctx context.Context, userID uuid.UUID, notificationID uuid.UUID) error {
result, err := s.pool.Exec(ctx, `
DELETE FROM notifications WHERE id = $1 AND user_id = $2
`, notificationID, userID)
if err != nil {
return err
}
if result.RowsAffected() == 0 {
return fmt.Errorf("notification not found")
}
return nil
}
// GetPreferences returns notification preferences for a user
func (s *NotificationService) GetPreferences(ctx context.Context, userID uuid.UUID) (*NotificationPreferences, error) {
var prefs NotificationPreferences
prefs.UserID = userID
err := s.pool.QueryRow(ctx, `
SELECT email_enabled, push_enabled, in_app_enabled, reminder_frequency
FROM notification_preferences
WHERE user_id = $1
`, userID).Scan(&prefs.EmailEnabled, &prefs.PushEnabled, &prefs.InAppEnabled, &prefs.ReminderFrequency)
if err != nil {
// Return defaults if not found
return &NotificationPreferences{
UserID: userID,
EmailEnabled: true,
PushEnabled: true,
InAppEnabled: true,
ReminderFrequency: "weekly",
}, nil
}
return &prefs, nil
}
// UpdatePreferences updates notification preferences for a user
func (s *NotificationService) UpdatePreferences(ctx context.Context, userID uuid.UUID, prefs *NotificationPreferences) error {
_, err := s.pool.Exec(ctx, `
INSERT INTO notification_preferences (user_id, email_enabled, push_enabled, in_app_enabled, reminder_frequency, updated_at)
VALUES ($1, $2, $3, $4, $5, NOW())
ON CONFLICT (user_id) DO UPDATE SET
email_enabled = $2,
push_enabled = $3,
in_app_enabled = $4,
reminder_frequency = $5,
updated_at = NOW()
`, userID, prefs.EmailEnabled, prefs.PushEnabled, prefs.InAppEnabled, prefs.ReminderFrequency)
return err
}
// NotifyConsentRequired sends consent required notifications to all active users
func (s *NotificationService) NotifyConsentRequired(ctx context.Context, documentName, versionID string) error {
// Get all active users
rows, err := s.pool.Query(ctx, `
SELECT id FROM users WHERE account_status = 'active'
`)
if err != nil {
return err
}
defer rows.Close()
title := "Neue Zustimmung erforderlich"
body := fmt.Sprintf("Eine neue Version von '%s' wurde veröffentlicht. Bitte überprüfen und bestätigen Sie diese.", documentName)
data := map[string]interface{}{
"version_id": versionID,
"document_name": documentName,
}
for rows.Next() {
var userID uuid.UUID
if err := rows.Scan(&userID); err != nil {
continue
}
go s.CreateNotification(ctx, userID, NotificationTypeConsentRequired, title, body, data)
}
return nil
}
// NotifyVersionApproved notifies the creator that their version was approved
func (s *NotificationService) NotifyVersionApproved(ctx context.Context, creatorID uuid.UUID, documentName, versionNumber, approverEmail string) error {
title := "Version genehmigt"
body := fmt.Sprintf("Ihre Version %s von '%s' wurde von %s genehmigt und kann nun veröffentlicht werden.", versionNumber, documentName, approverEmail)
data := map[string]interface{}{
"document_name": documentName,
"version_number": versionNumber,
"approver": approverEmail,
}
return s.CreateNotification(ctx, creatorID, NotificationTypeVersionApproved, title, body, data)
}
// NotifyVersionRejected notifies the creator that their version was rejected
func (s *NotificationService) NotifyVersionRejected(ctx context.Context, creatorID uuid.UUID, documentName, versionNumber, reason, rejecterEmail string) error {
title := "Version abgelehnt"
body := fmt.Sprintf("Ihre Version %s von '%s' wurde von %s abgelehnt. Grund: %s", versionNumber, documentName, rejecterEmail, reason)
data := map[string]interface{}{
"document_name": documentName,
"version_number": versionNumber,
"rejecter": rejecterEmail,
"reason": reason,
}
return s.CreateNotification(ctx, creatorID, NotificationTypeVersionRejected, title, body, data)
}