package privacy import ( "context" "database/sql" "encoding/json" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/niels/banking-app/backend/internal/audit" "github.com/niels/banking-app/backend/internal/domain" ) type Repository struct { db *pgxpool.Pool audit *audit.Repository } type DashboardMetrics struct { InventoryAssets int64 `json:"inventory_assets"` ActiveRetentionRules int64 `json:"active_retention_rules"` OpenDSRRequests int64 `json:"open_dsr_requests"` OverdueDSRRequests int64 `json:"overdue_dsr_requests"` PublishedNotices int64 `json:"published_notices"` FailedAttestations int64 `json:"failed_attestations"` ActiveResidencyRules int64 `json:"active_residency_rules"` } type Dashboard struct { Metrics DashboardMetrics `json:"metrics"` Requests []DataSubjectRequest `json:"data_subject_requests"` Attestations []SecurityAttestation `json:"security_attestations"` Notices []PrivacyNotice `json:"privacy_notices"` } type DataInventoryItem struct { ID string `json:"id"` SystemName string `json:"system_name"` TableName string `json:"table_name"` FieldName string `json:"field_name"` DataCategory string `json:"data_category"` Classification string `json:"classification"` LawfulBasis string `json:"lawful_basis"` RetentionPolicy string `json:"retention_policy"` ResidencyScope string `json:"residency_scope"` EncryptedAtRest bool `json:"encrypted_at_rest"` EncryptedInTransit bool `json:"encrypted_in_transit"` Owner string `json:"owner"` Notes string `json:"notes"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } type RetentionPolicy struct { ID string `json:"id"` PolicyKey string `json:"policy_key"` DataCategory string `json:"data_category"` RetentionPeriodDays int `json:"retention_period_days"` AnonymizationStrategy string `json:"anonymization_strategy"` DeletionStrategy string `json:"deletion_strategy"` LegalHoldAllowed bool `json:"legal_hold_allowed"` Active bool `json:"active"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } type RetentionPolicyParams struct { PolicyKey string DataCategory string RetentionPeriodDays int AnonymizationStrategy string DeletionStrategy string LegalHoldAllowed bool Active bool } type DataSubjectRequest struct { ID string `json:"id"` UserID string `json:"user_id,omitempty"` RequesterEmail string `json:"requester_email"` RequestType string `json:"request_type"` Status string `json:"status"` VerificationStatus string `json:"verification_status"` Details string `json:"details,omitempty"` ResponsePayload json.RawMessage `json:"response_payload"` AssignedToAdminUserID string `json:"assigned_to_admin_user_id,omitempty"` DueAt time.Time `json:"due_at"` CompletedAt *time.Time `json:"completed_at,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } type DataSubjectRequestParams struct { UserID string RequesterEmail string RequestType string Details string AdminUserID string } type DataSubjectRequestUpdate struct { ID string Status string VerificationStatus string AssignedToAdminUserID string Details *string } type PrivacyNotice struct { ID string `json:"id"` NoticeType string `json:"notice_type"` Version string `json:"version"` Status string `json:"status"` Content string `json:"content"` EffectiveAt *time.Time `json:"effective_at,omitempty"` PublishedByAdminUserID string `json:"published_by_admin_user_id,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } type PrivacyNoticeParams struct { NoticeType string Version string Status string Content string EffectiveAt *time.Time AdminUserID string } type SecurityAttestation struct { ID string `json:"id"` ControlType string `json:"control_type"` Status string `json:"status"` Evidence string `json:"evidence,omitempty"` CheckedByAdminUserID string `json:"checked_by_admin_user_id,omitempty"` CheckedAt time.Time `json:"checked_at"` CreatedAt time.Time `json:"created_at"` } type SecurityAttestationParams struct { ControlType string Status string Evidence string AdminUserID string } type ResidencyPolicy struct { ID string `json:"id"` RegionCode string `json:"region_code"` Status string `json:"status"` AllowedStorageRegions json.RawMessage `json:"allowed_storage_regions"` RestrictedDataCategories json.RawMessage `json:"restricted_data_categories"` Notes string `json:"notes,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } type ResidencyPolicyParams struct { RegionCode string Status string AllowedStorageRegions []string RestrictedDataCategories []string Notes string } type DeletionJob struct { ID string `json:"id"` DataSubjectRequestID string `json:"data_subject_request_id"` UserID string `json:"user_id,omitempty"` JobType string `json:"job_type"` Status string `json:"status"` AffectedTables json.RawMessage `json:"affected_tables"` Error string `json:"error,omitempty"` ProcessedAt *time.Time `json:"processed_at,omitempty"` CreatedAt time.Time `json:"created_at"` } func NewRepository(db *pgxpool.Pool, auditRepo *audit.Repository) *Repository { return &Repository{db: db, audit: auditRepo} } func (r *Repository) Dashboard(ctx context.Context, limit int) (Dashboard, error) { metrics, err := r.Metrics(ctx) if err != nil { return Dashboard{}, err } requests, err := r.ListDataSubjectRequests(ctx, "open", limit) if err != nil { return Dashboard{}, err } attestations, err := r.ListSecurityAttestations(ctx, "", limit) if err != nil { return Dashboard{}, err } notices, err := r.ListPrivacyNotices(ctx, "", false, limit) if err != nil { return Dashboard{}, err } return Dashboard{Metrics: metrics, Requests: requests, Attestations: attestations, Notices: notices}, nil } func (r *Repository) Metrics(ctx context.Context) (DashboardMetrics, error) { var metrics DashboardMetrics err := r.db.QueryRow(ctx, ` SELECT (SELECT COUNT(*) FROM privacy_data_inventory)::bigint, (SELECT COUNT(*) FROM data_retention_policies WHERE active = true)::bigint, (SELECT COUNT(*) FROM data_subject_requests WHERE status IN ('received', 'verifying_identity', 'in_progress'))::bigint, (SELECT COUNT(*) FROM data_subject_requests WHERE status IN ('received', 'verifying_identity', 'in_progress') AND due_at < now())::bigint, (SELECT COUNT(*) FROM privacy_notices WHERE status = 'published')::bigint, (SELECT COUNT(*) FROM data_protection_attestations dpa WHERE dpa.status = 'fail' AND dpa.checked_at = (SELECT max(last_dpa.checked_at) FROM data_protection_attestations last_dpa WHERE last_dpa.control_type = dpa.control_type))::bigint, (SELECT COUNT(*) FROM data_residency_policies WHERE status = 'active')::bigint `).Scan( &metrics.InventoryAssets, &metrics.ActiveRetentionRules, &metrics.OpenDSRRequests, &metrics.OverdueDSRRequests, &metrics.PublishedNotices, &metrics.FailedAttestations, &metrics.ActiveResidencyRules, ) return metrics, err } func (r *Repository) ListInventory(ctx context.Context, category, classification string, limit int) ([]DataInventoryItem, error) { limit = normalizeLimit(limit) rows, err := r.db.Query(ctx, ` SELECT id::text, system_name, table_name, field_name, data_category, classification, lawful_basis, retention_policy, residency_scope, encrypted_at_rest, encrypted_in_transit, owner, notes, created_at, updated_at FROM privacy_data_inventory WHERE (NULLIF($1, '') IS NULL OR data_category = $1) AND (NULLIF($2, '') IS NULL OR classification = $2) ORDER BY data_category, table_name, field_name LIMIT $3 `, category, classification, limit) if err != nil { return nil, err } defer rows.Close() items := []DataInventoryItem{} for rows.Next() { item, err := scanInventoryItem(rows) if err != nil { return nil, err } items = append(items, item) } return items, rows.Err() } func (r *Repository) ListRetentionPolicies(ctx context.Context, activeOnly bool, limit int) ([]RetentionPolicy, error) { limit = normalizeLimit(limit) rows, err := r.db.Query(ctx, ` SELECT id::text, policy_key, data_category, retention_period_days, anonymization_strategy, deletion_strategy, legal_hold_allowed, active, created_at, updated_at FROM data_retention_policies WHERE $1 = false OR active = true ORDER BY data_category, policy_key LIMIT $2 `, activeOnly, limit) if err != nil { return nil, err } defer rows.Close() policies := []RetentionPolicy{} for rows.Next() { policy, err := scanRetentionPolicy(rows) if err != nil { return nil, err } policies = append(policies, policy) } return policies, rows.Err() } func (r *Repository) UpsertRetentionPolicy(ctx context.Context, params RetentionPolicyParams) (RetentionPolicy, error) { row := r.db.QueryRow(ctx, ` INSERT INTO data_retention_policies ( policy_key, data_category, retention_period_days, anonymization_strategy, deletion_strategy, legal_hold_allowed, active ) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (policy_key) DO UPDATE SET data_category = EXCLUDED.data_category, retention_period_days = EXCLUDED.retention_period_days, anonymization_strategy = EXCLUDED.anonymization_strategy, deletion_strategy = EXCLUDED.deletion_strategy, legal_hold_allowed = EXCLUDED.legal_hold_allowed, active = EXCLUDED.active RETURNING id::text, policy_key, data_category, retention_period_days, anonymization_strategy, deletion_strategy, legal_hold_allowed, active, created_at, updated_at `, params.PolicyKey, params.DataCategory, params.RetentionPeriodDays, params.AnonymizationStrategy, params.DeletionStrategy, params.LegalHoldAllowed, params.Active) return scanRetentionPolicy(row) } func (r *Repository) CreateDataSubjectRequest(ctx context.Context, params DataSubjectRequestParams) (DataSubjectRequest, error) { userID := strings.TrimSpace(params.UserID) email := strings.ToLower(strings.TrimSpace(params.RequesterEmail)) if userID == "" && email != "" { _ = r.db.QueryRow(ctx, `SELECT id::text FROM users WHERE lower(email) = $1`, email).Scan(&userID) } dueAt := time.Now().UTC().Add(30 * 24 * time.Hour) row := r.db.QueryRow(ctx, ` INSERT INTO data_subject_requests ( user_id, requester_email, request_type, details, assigned_to_admin_user_id, due_at ) VALUES (NULLIF($1, '')::uuid, $2, $3, $4, NULLIF($5, '')::uuid, $6) RETURNING id::text, COALESCE(user_id::text, ''), requester_email, request_type, status, verification_status, details, response_payload, COALESCE(assigned_to_admin_user_id::text, ''), due_at, completed_at, created_at, updated_at `, userID, email, params.RequestType, params.Details, params.AdminUserID, dueAt) return scanDataSubjectRequest(row) } func (r *Repository) ListDataSubjectRequests(ctx context.Context, status string, limit int) ([]DataSubjectRequest, error) { limit = normalizeLimit(limit) status = strings.ToLower(strings.TrimSpace(status)) if status == "all" { status = "" } rows, err := r.db.Query(ctx, ` SELECT id::text, COALESCE(user_id::text, ''), requester_email, request_type, status, verification_status, details, response_payload, COALESCE(assigned_to_admin_user_id::text, ''), due_at, completed_at, created_at, updated_at FROM data_subject_requests WHERE NULLIF($1, '') IS NULL OR ($1 = 'open' AND status IN ('received', 'verifying_identity', 'in_progress')) OR status = $1 ORDER BY due_at ASC, created_at DESC LIMIT $2 `, status, limit) if err != nil { return nil, err } defer rows.Close() requests := []DataSubjectRequest{} for rows.Next() { request, err := scanDataSubjectRequest(rows) if err != nil { return nil, err } requests = append(requests, request) } return requests, rows.Err() } func (r *Repository) UpdateDataSubjectRequest(ctx context.Context, params DataSubjectRequestUpdate) (DataSubjectRequest, error) { row := r.db.QueryRow(ctx, ` UPDATE data_subject_requests SET status = CASE WHEN NULLIF($2, '') IS NULL THEN status ELSE $2 END, verification_status = CASE WHEN NULLIF($3, '') IS NULL THEN verification_status ELSE $3 END, assigned_to_admin_user_id = CASE WHEN $4::text IS NULL THEN assigned_to_admin_user_id ELSE NULLIF($4, '')::uuid END, details = CASE WHEN $5::text IS NULL THEN details ELSE $5 END, completed_at = CASE WHEN NULLIF($2, '') = 'completed' THEN now() ELSE completed_at END WHERE id = $1 RETURNING id::text, COALESCE(user_id::text, ''), requester_email, request_type, status, verification_status, details, response_payload, COALESCE(assigned_to_admin_user_id::text, ''), due_at, completed_at, created_at, updated_at `, params.ID, params.Status, params.VerificationStatus, emptyStringPtrValue(params.AssignedToAdminUserID), stringPtrValue(params.Details)) request, err := scanDataSubjectRequest(row) if errors.Is(err, pgx.ErrNoRows) { return DataSubjectRequest{}, domain.ErrNotFound } return request, err } func (r *Repository) ProcessDataSubjectRequest(ctx context.Context, requestID string, dryRun bool) (DataSubjectRequest, *DeletionJob, error) { tx, err := r.db.Begin(ctx) if err != nil { return DataSubjectRequest{}, nil, err } defer tx.Rollback(ctx) request, err := findDataSubjectRequestTx(ctx, tx, requestID) if errors.Is(err, pgx.ErrNoRows) { return DataSubjectRequest{}, nil, domain.ErrNotFound } if err != nil { return DataSubjectRequest{}, nil, err } if request.UserID == "" { return DataSubjectRequest{}, nil, fmt.Errorf("%w: data subject request is not linked to a user", domain.ErrValidation) } if request.RequestType == "erasure" && request.VerificationStatus != "verified" { return DataSubjectRequest{}, nil, fmt.Errorf("%w: erasure requires verified identity", domain.ErrValidation) } response := map[string]any{"request_type": request.RequestType, "dry_run": dryRun} affectedTables := []string{} var job *DeletionJob switch request.RequestType { case "access", "portability": summary, err := r.subjectAccessSummaryTx(ctx, tx, request.UserID) if err != nil { return DataSubjectRequest{}, nil, err } response["summary"] = summary case "erasure": affectedTables = []string{"users", "kyc_profiles", "kyc_documents", "beneficiaries", "payment_notifications"} if !dryRun { if err := anonymizeUserTx(ctx, tx, request.UserID); err != nil { return DataSubjectRequest{}, nil, err } } createdJob, err := insertDeletionJobTx(ctx, tx, request.ID, request.UserID, dryRun, "completed", affectedTables, "") if err != nil { return DataSubjectRequest{}, nil, err } job = &createdJob response["affected_tables"] = affectedTables default: response["workflow"] = "manual privacy workflow recorded; complete operational action outside automated anonymization" } raw, err := json.Marshal(response) if err != nil { return DataSubjectRequest{}, nil, err } status := "completed" if dryRun && request.RequestType == "erasure" { status = "in_progress" } row := tx.QueryRow(ctx, ` UPDATE data_subject_requests SET status = $2, response_payload = $3::jsonb, completed_at = CASE WHEN $2 = 'completed' THEN now() ELSE completed_at END WHERE id = $1 RETURNING id::text, COALESCE(user_id::text, ''), requester_email, request_type, status, verification_status, details, response_payload, COALESCE(assigned_to_admin_user_id::text, ''), due_at, completed_at, created_at, updated_at `, request.ID, status, string(raw)) updated, err := scanDataSubjectRequest(row) if err != nil { return DataSubjectRequest{}, nil, err } if err := tx.Commit(ctx); err != nil { return DataSubjectRequest{}, nil, err } return updated, job, nil } func (r *Repository) ListPrivacyNotices(ctx context.Context, noticeType string, publishedOnly bool, limit int) ([]PrivacyNotice, error) { limit = normalizeLimit(limit) rows, err := r.db.Query(ctx, ` SELECT id::text, notice_type, version, status, content, effective_at, COALESCE(published_by_admin_user_id::text, ''), created_at, updated_at FROM privacy_notices WHERE (NULLIF($1, '') IS NULL OR notice_type = $1) AND ($2 = false OR status = 'published') ORDER BY notice_type, effective_at DESC NULLS LAST, created_at DESC LIMIT $3 `, noticeType, publishedOnly, limit) if err != nil { return nil, err } defer rows.Close() notices := []PrivacyNotice{} for rows.Next() { notice, err := scanPrivacyNotice(rows) if err != nil { return nil, err } notices = append(notices, notice) } return notices, rows.Err() } func (r *Repository) UpsertPrivacyNotice(ctx context.Context, params PrivacyNoticeParams) (PrivacyNotice, error) { row := r.db.QueryRow(ctx, ` INSERT INTO privacy_notices (notice_type, version, status, content, effective_at, published_by_admin_user_id) VALUES ($1, $2, $3, $4, $5, NULLIF($6, '')::uuid) ON CONFLICT (notice_type, version) DO UPDATE SET status = EXCLUDED.status, content = EXCLUDED.content, effective_at = EXCLUDED.effective_at, published_by_admin_user_id = EXCLUDED.published_by_admin_user_id RETURNING id::text, notice_type, version, status, content, effective_at, COALESCE(published_by_admin_user_id::text, ''), created_at, updated_at `, params.NoticeType, params.Version, params.Status, params.Content, params.EffectiveAt, params.AdminUserID) return scanPrivacyNotice(row) } func (r *Repository) PublishPrivacyNotice(ctx context.Context, noticeID, adminUserID string) (PrivacyNotice, error) { tx, err := r.db.Begin(ctx) if err != nil { return PrivacyNotice{}, err } defer tx.Rollback(ctx) var noticeType string if err := tx.QueryRow(ctx, `SELECT notice_type FROM privacy_notices WHERE id = $1`, noticeID).Scan(¬iceType); err != nil { if errors.Is(err, pgx.ErrNoRows) { return PrivacyNotice{}, domain.ErrNotFound } return PrivacyNotice{}, err } if _, err := tx.Exec(ctx, `UPDATE privacy_notices SET status = 'retired' WHERE notice_type = $1 AND status = 'published'`, noticeType); err != nil { return PrivacyNotice{}, err } row := tx.QueryRow(ctx, ` UPDATE privacy_notices SET status = 'published', effective_at = COALESCE(effective_at, now()), published_by_admin_user_id = NULLIF($2, '')::uuid WHERE id = $1 RETURNING id::text, notice_type, version, status, content, effective_at, COALESCE(published_by_admin_user_id::text, ''), created_at, updated_at `, noticeID, adminUserID) notice, err := scanPrivacyNotice(row) if err != nil { return PrivacyNotice{}, err } if err := tx.Commit(ctx); err != nil { return PrivacyNotice{}, err } return notice, nil } func (r *Repository) ListSecurityAttestations(ctx context.Context, controlType string, limit int) ([]SecurityAttestation, error) { limit = normalizeLimit(limit) rows, err := r.db.Query(ctx, ` SELECT id::text, control_type, status, evidence, COALESCE(checked_by_admin_user_id::text, ''), checked_at, created_at FROM data_protection_attestations WHERE NULLIF($1, '') IS NULL OR control_type = $1 ORDER BY checked_at DESC LIMIT $2 `, controlType, limit) if err != nil { return nil, err } defer rows.Close() attestations := []SecurityAttestation{} for rows.Next() { attestation, err := scanSecurityAttestation(rows) if err != nil { return nil, err } attestations = append(attestations, attestation) } return attestations, rows.Err() } func (r *Repository) CreateSecurityAttestation(ctx context.Context, params SecurityAttestationParams) (SecurityAttestation, error) { row := r.db.QueryRow(ctx, ` INSERT INTO data_protection_attestations (control_type, status, evidence, checked_by_admin_user_id) VALUES ($1, $2, $3, NULLIF($4, '')::uuid) RETURNING id::text, control_type, status, evidence, COALESCE(checked_by_admin_user_id::text, ''), checked_at, created_at `, params.ControlType, params.Status, params.Evidence, params.AdminUserID) return scanSecurityAttestation(row) } func (r *Repository) ListResidencyPolicies(ctx context.Context, limit int) ([]ResidencyPolicy, error) { limit = normalizeLimit(limit) rows, err := r.db.Query(ctx, ` SELECT id::text, region_code, status, allowed_storage_regions, restricted_data_categories, notes, created_at, updated_at FROM data_residency_policies ORDER BY region_code LIMIT $1 `, limit) if err != nil { return nil, err } defer rows.Close() policies := []ResidencyPolicy{} for rows.Next() { policy, err := scanResidencyPolicy(rows) if err != nil { return nil, err } policies = append(policies, policy) } return policies, rows.Err() } func (r *Repository) UpsertResidencyPolicy(ctx context.Context, params ResidencyPolicyParams) (ResidencyPolicy, error) { allowed, err := json.Marshal(params.AllowedStorageRegions) if err != nil { return ResidencyPolicy{}, err } restricted, err := json.Marshal(params.RestrictedDataCategories) if err != nil { return ResidencyPolicy{}, err } row := r.db.QueryRow(ctx, ` INSERT INTO data_residency_policies (region_code, status, allowed_storage_regions, restricted_data_categories, notes) VALUES ($1, $2, $3::jsonb, $4::jsonb, $5) ON CONFLICT (region_code) DO UPDATE SET status = EXCLUDED.status, allowed_storage_regions = EXCLUDED.allowed_storage_regions, restricted_data_categories = EXCLUDED.restricted_data_categories, notes = EXCLUDED.notes RETURNING id::text, region_code, status, allowed_storage_regions, restricted_data_categories, notes, created_at, updated_at `, params.RegionCode, params.Status, string(allowed), string(restricted), params.Notes) return scanResidencyPolicy(row) } func (r *Repository) VerifyAuditIntegrity(ctx context.Context, limit int) (audit.IntegrityReport, error) { return r.audit.VerifyIntegrity(ctx, limit) } func scanInventoryItem(row interface{ Scan(dest ...any) error }) (DataInventoryItem, error) { var item DataInventoryItem err := row.Scan(&item.ID, &item.SystemName, &item.TableName, &item.FieldName, &item.DataCategory, &item.Classification, &item.LawfulBasis, &item.RetentionPolicy, &item.ResidencyScope, &item.EncryptedAtRest, &item.EncryptedInTransit, &item.Owner, &item.Notes, &item.CreatedAt, &item.UpdatedAt) return item, err } func scanRetentionPolicy(row interface{ Scan(dest ...any) error }) (RetentionPolicy, error) { var policy RetentionPolicy err := row.Scan(&policy.ID, &policy.PolicyKey, &policy.DataCategory, &policy.RetentionPeriodDays, &policy.AnonymizationStrategy, &policy.DeletionStrategy, &policy.LegalHoldAllowed, &policy.Active, &policy.CreatedAt, &policy.UpdatedAt) return policy, err } func scanDataSubjectRequest(row interface{ Scan(dest ...any) error }) (DataSubjectRequest, error) { var request DataSubjectRequest var completedAt sql.NullTime err := row.Scan(&request.ID, &request.UserID, &request.RequesterEmail, &request.RequestType, &request.Status, &request.VerificationStatus, &request.Details, &request.ResponsePayload, &request.AssignedToAdminUserID, &request.DueAt, &completedAt, &request.CreatedAt, &request.UpdatedAt) if completedAt.Valid { request.CompletedAt = &completedAt.Time } return request, err } func scanPrivacyNotice(row interface{ Scan(dest ...any) error }) (PrivacyNotice, error) { var notice PrivacyNotice var effectiveAt sql.NullTime err := row.Scan(¬ice.ID, ¬ice.NoticeType, ¬ice.Version, ¬ice.Status, ¬ice.Content, &effectiveAt, ¬ice.PublishedByAdminUserID, ¬ice.CreatedAt, ¬ice.UpdatedAt) if effectiveAt.Valid { notice.EffectiveAt = &effectiveAt.Time } return notice, err } func scanSecurityAttestation(row interface{ Scan(dest ...any) error }) (SecurityAttestation, error) { var attestation SecurityAttestation err := row.Scan(&attestation.ID, &attestation.ControlType, &attestation.Status, &attestation.Evidence, &attestation.CheckedByAdminUserID, &attestation.CheckedAt, &attestation.CreatedAt) return attestation, err } func scanResidencyPolicy(row interface{ Scan(dest ...any) error }) (ResidencyPolicy, error) { var policy ResidencyPolicy err := row.Scan(&policy.ID, &policy.RegionCode, &policy.Status, &policy.AllowedStorageRegions, &policy.RestrictedDataCategories, &policy.Notes, &policy.CreatedAt, &policy.UpdatedAt) return policy, err } func scanDeletionJob(row interface{ Scan(dest ...any) error }) (DeletionJob, error) { var job DeletionJob var processedAt sql.NullTime err := row.Scan(&job.ID, &job.DataSubjectRequestID, &job.UserID, &job.JobType, &job.Status, &job.AffectedTables, &job.Error, &processedAt, &job.CreatedAt) if processedAt.Valid { job.ProcessedAt = &processedAt.Time } return job, err } func findDataSubjectRequestTx(ctx context.Context, tx pgx.Tx, id string) (DataSubjectRequest, error) { row := tx.QueryRow(ctx, ` SELECT id::text, COALESCE(user_id::text, ''), requester_email, request_type, status, verification_status, details, response_payload, COALESCE(assigned_to_admin_user_id::text, ''), due_at, completed_at, created_at, updated_at FROM data_subject_requests WHERE id = $1 FOR UPDATE `, id) return scanDataSubjectRequest(row) } func (r *Repository) subjectAccessSummaryTx(ctx context.Context, tx pgx.Tx, userID string) (map[string]any, error) { var email, fullName, kycStatus string var accountCount, walletCount, transferCount, amlCaseCount, auditCount int64 err := tx.QueryRow(ctx, ` SELECT u.email, u.full_name, COALESCE((SELECT kp.status FROM kyc_profiles kp WHERE kp.user_id = u.id), 'not_started'), (SELECT COUNT(*) FROM accounts WHERE user_id = u.id)::bigint, (SELECT COUNT(*) FROM wallets WHERE user_id = u.id)::bigint, (SELECT COUNT(*) FROM transfers WHERE user_id = u.id)::bigint, (SELECT COUNT(*) FROM aml_cases WHERE user_id = u.id)::bigint, (SELECT COUNT(*) FROM audit_events WHERE actor_user_id = u.id)::bigint FROM users u WHERE u.id = $1 `, userID).Scan(&email, &fullName, &kycStatus, &accountCount, &walletCount, &transferCount, &amlCaseCount, &auditCount) if errors.Is(err, pgx.ErrNoRows) { return nil, domain.ErrNotFound } if err != nil { return nil, err } return map[string]any{ "user": map[string]any{ "id": userID, "email": email, "full_name": fullName, "kyc_status": kycStatus, }, "counts": map[string]any{ "accounts": accountCount, "wallets": walletCount, "transfers": transferCount, "aml_cases": amlCaseCount, "audit_events": auditCount, }, }, nil } func anonymizeUserTx(ctx context.Context, tx pgx.Tx, userID string) error { suffix := strings.ReplaceAll(userID, "-", "") if len(suffix) > 12 { suffix = suffix[:12] } anonymizedEmail := "anonymized+" + suffix + "@example.invalid" updates := []struct { sql string args []any }{ {`UPDATE users SET email = $2, full_name = 'Anonymized User' WHERE id = $1`, []any{userID, anonymizedEmail}}, {`UPDATE kyc_profiles SET legal_name = 'Anonymized User', address_line1 = 'Anonymized', city = 'Anonymized', postal_code = '0000AA', rejection_reason = NULL WHERE user_id = $1`, []any{userID}}, {`UPDATE kyc_documents SET document_number = NULL WHERE kyc_profile_id IN (SELECT id FROM kyc_profiles WHERE user_id = $1)`, []any{userID}}, {`UPDATE beneficiaries SET name = 'Anonymized Beneficiary', iban = 'ANONYMIZED-' || left(id::text, 12), bic = '' WHERE user_id = $1`, []any{userID}}, {`UPDATE transfers SET beneficiary_name = 'Anonymized Beneficiary', beneficiary_iban = 'ANONYMIZED-' || left(id::text, 12), beneficiary_bic = '' WHERE user_id = $1 AND beneficiary_id IS NOT NULL`, []any{userID}}, {`UPDATE payment_notifications SET title = 'Privacy request processed', message = 'Notification content anonymized', payload = '{}'::jsonb WHERE user_id = $1`, []any{userID}}, } for _, update := range updates { if _, err := tx.Exec(ctx, update.sql, update.args...); err != nil { return err } } return nil } func insertDeletionJobTx(ctx context.Context, tx pgx.Tx, requestID, userID string, dryRun bool, status string, affectedTables []string, errText string) (DeletionJob, error) { jobType := "anonymize" if dryRun { jobType = "dry_run" } rawTables, err := json.Marshal(affectedTables) if err != nil { return DeletionJob{}, err } row := tx.QueryRow(ctx, ` INSERT INTO privacy_deletion_jobs ( data_subject_request_id, user_id, job_type, status, affected_tables, error, processed_at ) VALUES ($1, NULLIF($2, '')::uuid, $3, $4, $5::jsonb, $6, now()) RETURNING id::text, data_subject_request_id::text, COALESCE(user_id::text, ''), job_type, status, affected_tables, error, processed_at, created_at `, requestID, userID, jobType, status, string(rawTables), errText) return scanDeletionJob(row) } func normalizeLimit(limit int) int { if limit <= 0 || limit > 100 { return 50 } return limit } func stringPtrValue(value *string) any { if value == nil { return nil } return *value } func emptyStringPtrValue(value string) any { if strings.TrimSpace(value) == "" { return nil } return strings.TrimSpace(value) }