package services import ( "context" "encoding/json" "fmt" "time" "github.com/breakpilot/consent-service/internal/models" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) // DSRService handles Data Subject Request business logic type DSRService struct { pool *pgxpool.Pool notificationService *NotificationService emailService *EmailService } // NewDSRService creates a new DSRService func NewDSRService(pool *pgxpool.Pool, notificationService *NotificationService, emailService *EmailService) *DSRService { return &DSRService{ pool: pool, notificationService: notificationService, emailService: emailService, } } // GetPool returns the database pool for direct queries func (s *DSRService) GetPool() *pgxpool.Pool { return s.pool } // generateRequestNumber generates a unique request number like DSR-2025-000001 func (s *DSRService) generateRequestNumber(ctx context.Context) (string, error) { var seqNum int64 err := s.pool.QueryRow(ctx, "SELECT nextval('dsr_request_number_seq')").Scan(&seqNum) if err != nil { return "", fmt.Errorf("failed to get next sequence number: %w", err) } year := time.Now().Year() return fmt.Sprintf("DSR-%d-%06d", year, seqNum), nil } // CreateRequest creates a new data subject request func (s *DSRService) CreateRequest(ctx context.Context, req models.CreateDSRRequest, createdBy *uuid.UUID) (*models.DataSubjectRequest, error) { // Validate request type requestType := models.DSRRequestType(req.RequestType) if !isValidRequestType(requestType) { return nil, fmt.Errorf("invalid request type: %s", req.RequestType) } // Generate request number requestNumber, err := s.generateRequestNumber(ctx) if err != nil { return nil, err } // Calculate deadline deadlineDays := requestType.DeadlineDays() deadline := time.Now().AddDate(0, 0, deadlineDays) // Determine priority priority := models.DSRPriorityNormal if req.Priority != "" { priority = models.DSRPriority(req.Priority) } else if requestType.IsExpedited() { priority = models.DSRPriorityExpedited } // Determine source source := models.DSRSourceAPI if req.Source != "" { source = models.DSRSource(req.Source) } // Serialize request details detailsJSON, err := json.Marshal(req.RequestDetails) if err != nil { detailsJSON = []byte("{}") } // Try to find existing user by email var userID *uuid.UUID var foundUserID uuid.UUID err = s.pool.QueryRow(ctx, "SELECT id FROM users WHERE email = $1", req.RequesterEmail).Scan(&foundUserID) if err == nil { userID = &foundUserID } // Insert request var dsr models.DataSubjectRequest err = s.pool.QueryRow(ctx, ` INSERT INTO data_subject_requests ( user_id, request_number, request_type, status, priority, source, requester_email, requester_name, requester_phone, request_details, deadline_at, legal_deadline_days, created_by ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING id, user_id, request_number, request_type, status, priority, source, requester_email, requester_name, requester_phone, identity_verified, request_details, deadline_at, legal_deadline_days, created_at, updated_at, created_by `, userID, requestNumber, requestType, models.DSRStatusIntake, priority, source, req.RequesterEmail, req.RequesterName, req.RequesterPhone, detailsJSON, deadline, deadlineDays, createdBy, ).Scan( &dsr.ID, &dsr.UserID, &dsr.RequestNumber, &dsr.RequestType, &dsr.Status, &dsr.Priority, &dsr.Source, &dsr.RequesterEmail, &dsr.RequesterName, &dsr.RequesterPhone, &dsr.IdentityVerified, &detailsJSON, &dsr.DeadlineAt, &dsr.LegalDeadlineDays, &dsr.CreatedAt, &dsr.UpdatedAt, &dsr.CreatedBy, ) if err != nil { return nil, fmt.Errorf("failed to create DSR: %w", err) } // Parse details back json.Unmarshal(detailsJSON, &dsr.RequestDetails) // Record initial status s.recordStatusChange(ctx, dsr.ID, nil, models.DSRStatusIntake, createdBy, "Anfrage eingegangen") // Notify DPOs about new request go s.notifyNewRequest(context.Background(), &dsr) return &dsr, nil } // GetByID retrieves a DSR by ID func (s *DSRService) GetByID(ctx context.Context, id uuid.UUID) (*models.DataSubjectRequest, error) { var dsr models.DataSubjectRequest var detailsJSON, resultDataJSON []byte err := s.pool.QueryRow(ctx, ` SELECT id, user_id, request_number, request_type, status, priority, source, requester_email, requester_name, requester_phone, identity_verified, identity_verified_at, identity_verified_by, identity_verification_method, request_details, deadline_at, legal_deadline_days, extended_deadline_at, extension_reason, assigned_to, processing_notes, completed_at, completed_by, result_summary, result_data, rejected_at, rejected_by, rejection_reason, rejection_legal_basis, created_at, updated_at, created_by FROM data_subject_requests WHERE id = $1 `, id).Scan( &dsr.ID, &dsr.UserID, &dsr.RequestNumber, &dsr.RequestType, &dsr.Status, &dsr.Priority, &dsr.Source, &dsr.RequesterEmail, &dsr.RequesterName, &dsr.RequesterPhone, &dsr.IdentityVerified, &dsr.IdentityVerifiedAt, &dsr.IdentityVerifiedBy, &dsr.IdentityVerificationMethod, &detailsJSON, &dsr.DeadlineAt, &dsr.LegalDeadlineDays, &dsr.ExtendedDeadlineAt, &dsr.ExtensionReason, &dsr.AssignedTo, &dsr.ProcessingNotes, &dsr.CompletedAt, &dsr.CompletedBy, &dsr.ResultSummary, &resultDataJSON, &dsr.RejectedAt, &dsr.RejectedBy, &dsr.RejectionReason, &dsr.RejectionLegalBasis, &dsr.CreatedAt, &dsr.UpdatedAt, &dsr.CreatedBy, ) if err != nil { return nil, fmt.Errorf("DSR not found: %w", err) } json.Unmarshal(detailsJSON, &dsr.RequestDetails) json.Unmarshal(resultDataJSON, &dsr.ResultData) return &dsr, nil } // GetByNumber retrieves a DSR by request number func (s *DSRService) GetByNumber(ctx context.Context, requestNumber string) (*models.DataSubjectRequest, error) { var id uuid.UUID err := s.pool.QueryRow(ctx, "SELECT id FROM data_subject_requests WHERE request_number = $1", requestNumber).Scan(&id) if err != nil { return nil, fmt.Errorf("DSR not found: %w", err) } return s.GetByID(ctx, id) } // List retrieves DSRs with filters and pagination func (s *DSRService) List(ctx context.Context, filters models.DSRListFilters, limit, offset int) ([]models.DataSubjectRequest, int, error) { // Build query baseQuery := "FROM data_subject_requests WHERE 1=1" args := []interface{}{} argIndex := 1 if filters.Status != nil && *filters.Status != "" { baseQuery += fmt.Sprintf(" AND status = $%d", argIndex) args = append(args, *filters.Status) argIndex++ } if filters.RequestType != nil && *filters.RequestType != "" { baseQuery += fmt.Sprintf(" AND request_type = $%d", argIndex) args = append(args, *filters.RequestType) argIndex++ } if filters.AssignedTo != nil && *filters.AssignedTo != "" { baseQuery += fmt.Sprintf(" AND assigned_to = $%d", argIndex) args = append(args, *filters.AssignedTo) argIndex++ } if filters.Priority != nil && *filters.Priority != "" { baseQuery += fmt.Sprintf(" AND priority = $%d", argIndex) args = append(args, *filters.Priority) argIndex++ } if filters.OverdueOnly { baseQuery += " AND deadline_at < NOW() AND status NOT IN ('completed', 'rejected', 'cancelled')" } if filters.FromDate != nil { baseQuery += fmt.Sprintf(" AND created_at >= $%d", argIndex) args = append(args, *filters.FromDate) argIndex++ } if filters.ToDate != nil { baseQuery += fmt.Sprintf(" AND created_at <= $%d", argIndex) args = append(args, *filters.ToDate) argIndex++ } if filters.Search != nil && *filters.Search != "" { searchPattern := "%" + *filters.Search + "%" baseQuery += fmt.Sprintf(" AND (request_number ILIKE $%d OR requester_email ILIKE $%d OR requester_name ILIKE $%d)", argIndex, argIndex, argIndex) args = append(args, searchPattern) argIndex++ } // Get total count var total int err := s.pool.QueryRow(ctx, "SELECT COUNT(*) "+baseQuery, args...).Scan(&total) if err != nil { return nil, 0, fmt.Errorf("failed to count DSRs: %w", err) } // Get paginated results query := fmt.Sprintf(` SELECT id, user_id, request_number, request_type, status, priority, source, requester_email, requester_name, requester_phone, identity_verified, deadline_at, legal_deadline_days, assigned_to, created_at, updated_at %s ORDER BY created_at DESC LIMIT $%d OFFSET $%d `, baseQuery, argIndex, argIndex+1) args = append(args, limit, offset) rows, err := s.pool.Query(ctx, query, args...) if err != nil { return nil, 0, fmt.Errorf("failed to query DSRs: %w", err) } defer rows.Close() var dsrs []models.DataSubjectRequest for rows.Next() { var dsr models.DataSubjectRequest err := rows.Scan( &dsr.ID, &dsr.UserID, &dsr.RequestNumber, &dsr.RequestType, &dsr.Status, &dsr.Priority, &dsr.Source, &dsr.RequesterEmail, &dsr.RequesterName, &dsr.RequesterPhone, &dsr.IdentityVerified, &dsr.DeadlineAt, &dsr.LegalDeadlineDays, &dsr.AssignedTo, &dsr.CreatedAt, &dsr.UpdatedAt, ) if err != nil { return nil, 0, fmt.Errorf("failed to scan DSR: %w", err) } dsrs = append(dsrs, dsr) } return dsrs, total, nil } // ListByUser retrieves DSRs for a specific user func (s *DSRService) ListByUser(ctx context.Context, userID uuid.UUID) ([]models.DataSubjectRequest, error) { rows, err := s.pool.Query(ctx, ` SELECT id, user_id, request_number, request_type, status, priority, source, requester_email, requester_name, deadline_at, created_at, updated_at FROM data_subject_requests WHERE user_id = $1 OR requester_email = (SELECT email FROM users WHERE id = $1) ORDER BY created_at DESC `, userID) if err != nil { return nil, fmt.Errorf("failed to query user DSRs: %w", err) } defer rows.Close() var dsrs []models.DataSubjectRequest for rows.Next() { var dsr models.DataSubjectRequest err := rows.Scan( &dsr.ID, &dsr.UserID, &dsr.RequestNumber, &dsr.RequestType, &dsr.Status, &dsr.Priority, &dsr.Source, &dsr.RequesterEmail, &dsr.RequesterName, &dsr.DeadlineAt, &dsr.CreatedAt, &dsr.UpdatedAt, ) if err != nil { return nil, fmt.Errorf("failed to scan DSR: %w", err) } dsrs = append(dsrs, dsr) } return dsrs, nil } // UpdateStatus changes the status of a DSR func (s *DSRService) UpdateStatus(ctx context.Context, id uuid.UUID, newStatus models.DSRStatus, comment string, changedBy *uuid.UUID) error { // Get current status var currentStatus models.DSRStatus err := s.pool.QueryRow(ctx, "SELECT status FROM data_subject_requests WHERE id = $1", id).Scan(¤tStatus) if err != nil { return fmt.Errorf("DSR not found: %w", err) } // Validate transition if !isValidStatusTransition(currentStatus, newStatus) { return fmt.Errorf("invalid status transition from %s to %s", currentStatus, newStatus) } // Update status _, err = s.pool.Exec(ctx, ` UPDATE data_subject_requests SET status = $1, updated_at = NOW() WHERE id = $2 `, newStatus, id) if err != nil { return fmt.Errorf("failed to update status: %w", err) } // Record status change s.recordStatusChange(ctx, id, ¤tStatus, newStatus, changedBy, comment) return nil } // VerifyIdentity marks identity as verified func (s *DSRService) VerifyIdentity(ctx context.Context, id uuid.UUID, method string, verifiedBy uuid.UUID) error { _, err := s.pool.Exec(ctx, ` UPDATE data_subject_requests SET identity_verified = TRUE, identity_verified_at = NOW(), identity_verified_by = $1, identity_verification_method = $2, status = CASE WHEN status = 'intake' THEN 'identity_verification' ELSE status END, updated_at = NOW() WHERE id = $3 `, verifiedBy, method, id) if err != nil { return fmt.Errorf("failed to verify identity: %w", err) } s.recordStatusChange(ctx, id, nil, models.DSRStatusIdentityVerification, &verifiedBy, "Identität verifiziert via "+method) return nil } // AssignRequest assigns a DSR to a handler func (s *DSRService) AssignRequest(ctx context.Context, id uuid.UUID, assigneeID uuid.UUID, assignedBy uuid.UUID) error { _, err := s.pool.Exec(ctx, ` UPDATE data_subject_requests SET assigned_to = $1, updated_at = NOW() WHERE id = $2 `, assigneeID, id) if err != nil { return fmt.Errorf("failed to assign DSR: %w", err) } // Get assignee name for comment var assigneeName string s.pool.QueryRow(ctx, "SELECT COALESCE(name, email) FROM users WHERE id = $1", assigneeID).Scan(&assigneeName) s.recordStatusChange(ctx, id, nil, "", &assignedBy, "Zugewiesen an "+assigneeName) // Notify assignee go s.notifyAssignment(context.Background(), id, assigneeID) return nil } // CompleteRequest marks a DSR as completed func (s *DSRService) CompleteRequest(ctx context.Context, id uuid.UUID, summary string, resultData map[string]interface{}, completedBy uuid.UUID) error { resultJSON, _ := json.Marshal(resultData) // Get current status var currentStatus models.DSRStatus s.pool.QueryRow(ctx, "SELECT status FROM data_subject_requests WHERE id = $1", id).Scan(¤tStatus) _, err := s.pool.Exec(ctx, ` UPDATE data_subject_requests SET status = 'completed', completed_at = NOW(), completed_by = $1, result_summary = $2, result_data = $3, updated_at = NOW() WHERE id = $4 `, completedBy, summary, resultJSON, id) if err != nil { return fmt.Errorf("failed to complete DSR: %w", err) } s.recordStatusChange(ctx, id, ¤tStatus, models.DSRStatusCompleted, &completedBy, summary) return nil } // RejectRequest rejects a DSR with legal basis func (s *DSRService) RejectRequest(ctx context.Context, id uuid.UUID, reason, legalBasis string, rejectedBy uuid.UUID) error { // Get current status var currentStatus models.DSRStatus s.pool.QueryRow(ctx, "SELECT status FROM data_subject_requests WHERE id = $1", id).Scan(¤tStatus) _, err := s.pool.Exec(ctx, ` UPDATE data_subject_requests SET status = 'rejected', rejected_at = NOW(), rejected_by = $1, rejection_reason = $2, rejection_legal_basis = $3, updated_at = NOW() WHERE id = $4 `, rejectedBy, reason, legalBasis, id) if err != nil { return fmt.Errorf("failed to reject DSR: %w", err) } s.recordStatusChange(ctx, id, ¤tStatus, models.DSRStatusRejected, &rejectedBy, fmt.Sprintf("Abgelehnt (%s): %s", legalBasis, reason)) return nil } // CancelRequest cancels a DSR (by user) func (s *DSRService) CancelRequest(ctx context.Context, id uuid.UUID, cancelledBy uuid.UUID) error { // Verify ownership var userID *uuid.UUID err := s.pool.QueryRow(ctx, "SELECT user_id FROM data_subject_requests WHERE id = $1", id).Scan(&userID) if err != nil { return fmt.Errorf("DSR not found: %w", err) } if userID == nil || *userID != cancelledBy { return fmt.Errorf("unauthorized: can only cancel own requests") } // Get current status var currentStatus models.DSRStatus s.pool.QueryRow(ctx, "SELECT status FROM data_subject_requests WHERE id = $1", id).Scan(¤tStatus) _, err = s.pool.Exec(ctx, ` UPDATE data_subject_requests SET status = 'cancelled', updated_at = NOW() WHERE id = $1 `, id) if err != nil { return fmt.Errorf("failed to cancel DSR: %w", err) } s.recordStatusChange(ctx, id, ¤tStatus, models.DSRStatusCancelled, &cancelledBy, "Vom Antragsteller storniert") return nil } // --- Internal helpers --- func (s *DSRService) recordStatusChange(ctx context.Context, requestID uuid.UUID, fromStatus *models.DSRStatus, toStatus models.DSRStatus, changedBy *uuid.UUID, comment string) { s.pool.Exec(ctx, ` INSERT INTO dsr_status_history (request_id, from_status, to_status, changed_by, comment) VALUES ($1, $2, $3, $4, $5) `, requestID, fromStatus, toStatus, changedBy, comment) } func isValidRequestType(rt models.DSRRequestType) bool { switch rt { case models.DSRTypeAccess, models.DSRTypeRectification, models.DSRTypeErasure, models.DSRTypeRestriction, models.DSRTypePortability: return true } return false } func isValidStatusTransition(from, to models.DSRStatus) bool { validTransitions := map[models.DSRStatus][]models.DSRStatus{ models.DSRStatusIntake: {models.DSRStatusIdentityVerification, models.DSRStatusProcessing, models.DSRStatusRejected, models.DSRStatusCancelled}, models.DSRStatusIdentityVerification: {models.DSRStatusProcessing, models.DSRStatusRejected, models.DSRStatusCancelled}, models.DSRStatusProcessing: {models.DSRStatusCompleted, models.DSRStatusRejected, models.DSRStatusCancelled}, models.DSRStatusCompleted: {}, models.DSRStatusRejected: {}, models.DSRStatusCancelled: {}, } allowed, exists := validTransitions[from] if !exists { return false } for _, s := range allowed { if s == to { return true } } return false }