package saas import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" "data-terminal-cloud/internal/audit" "data-terminal-cloud/internal/database" "data-terminal-cloud/internal/identity" "data-terminal-cloud/internal/outbox" "github.com/jackc/pgx/v5" ) var ErrIdempotencyConflict = errors.New("idempotency key was reused with different request data") type PrivacySubtask struct { ID string `json:"id"` Subsystem string `json:"subsystem"` TargetType string `json:"target_type"` TargetRef string `json:"target_ref"` Status string `json:"status"` Attempts int `json:"attempts"` NextAttemptAt *time.Time `json:"next_attempt_at,omitempty"` LastError string `json:"last_error,omitempty"` CompletedAt *time.Time `json:"completed_at,omitempty"` } type PrivacyRequest struct { ID string `json:"id"` SubjectID string `json:"subject_id"` RequestType string `json:"request_type"` Scope map[string]any `json:"scope"` Status string `json:"status"` DeadlineAt *time.Time `json:"deadline_at,omitempty"` LegalHold bool `json:"legal_hold"` ResultSummary map[string]any `json:"result_summary"` Version int64 `json:"version"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` CompletedAt *time.Time `json:"completed_at,omitempty"` Subtasks []PrivacySubtask `json:"subtasks"` } type CreatePrivacyRequestInput struct { RequestType string `json:"request_type"` Scope map[string]any `json:"scope"` Reason string `json:"-"` Confirmation string `json:"-"` } func (s *Service) CreatePrivacyRequest(ctx context.Context, principal identity.Principal, subjectPublicID, idempotencyKey, requestID string, input CreatePrivacyRequestInput) (PrivacyRequest, bool, error) { input.RequestType = strings.ToUpper(strings.TrimSpace(input.RequestType)) input.Reason = strings.TrimSpace(input.Reason) input.Confirmation = strings.TrimSpace(input.Confirmation) if input.Scope == nil { input.Scope = map[string]any{} } if input.RequestType != "ACCESS" && input.RequestType != "DELETE" { return PrivacyRequest{}, false, ErrNotFound } if input.RequestType == "DELETE" { reasonDigest := sha256.Sum256([]byte(input.Reason)) input.Scope = map[string]any{"reason_digest": hex.EncodeToString(reasonDigest[:])} } confirmationDigest := "" if input.RequestType == "DELETE" { value := sha256.Sum256([]byte(input.Confirmation)) confirmationDigest = hex.EncodeToString(value[:]) } requestBytes, err := json.Marshal(struct { RequestType string `json:"request_type"` Scope map[string]any `json:"scope"` ConfirmationDigest string `json:"confirmation_digest,omitempty"` }{input.RequestType, input.Scope, confirmationDigest}) if err != nil { return PrivacyRequest{}, false, fmt.Errorf("encode privacy request: %w", err) } digestValue := sha256.Sum256(requestBytes) digest := hex.EncodeToString(digestValue[:]) created := false var publicID string err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error { // Idempotent replay remains readable after the first DELETE request has // frozen the subject and removed it from normal resource resolution. var replayDigest string replayErr := tx.QueryRow(ctx, ` SELECT pr.public_id::text,pr.request_digest FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id AND sp.tenant_id=pr.tenant_id JOIN household_member hm ON hm.household_id=sp.household_id AND hm.tenant_id=sp.tenant_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=sp.tenant_id AND ua.status='active' WHERE sp.public_id=$1::uuid AND pr.idempotency_key=$2 AND ua.oidc_issuer=$3 AND ua.oidc_subject=$4 AND pr.requested_by=ua.id`, subjectPublicID, idempotencyKey, principal.Issuer, principal.Subject).Scan(&publicID, &replayDigest) if replayErr == nil { if replayDigest != digest { return ErrIdempotencyConflict } return nil } if !errors.Is(replayErr, pgx.ErrNoRows) { return replayErr } access, subject, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilityPrivacyManage, true) if err != nil { return err } if input.RequestType == "DELETE" { if input.Confirmation != subject.PreferredName { return ErrDeleteConfirmation } var activeCount int if err := tx.QueryRow(ctx, `SELECT count(*) FROM conversation WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')`, access.tenantID, access.subjectID).Scan(&activeCount); err != nil { return err } if activeCount > 0 { return ErrSubjectDeleteConflict } } scopeJSON, err := json.Marshal(input.Scope) if err != nil { return err } var internalID string err = tx.QueryRow(ctx, ` INSERT INTO privacy_request(tenant_id,subject_id,request_type,scope,requested_by,idempotency_key,request_digest) VALUES($1::uuid,$2::uuid,$3,$4::jsonb,$5::uuid,$6,$7) ON CONFLICT (tenant_id,requested_by,idempotency_key) DO NOTHING RETURNING id::text,public_id::text`, access.tenantID, access.subjectID, input.RequestType, scopeJSON, access.userID, idempotencyKey, digest, ).Scan(&internalID, &publicID) if errors.Is(err, pgx.ErrNoRows) { var storedDigest string if err := tx.QueryRow(ctx, `SELECT public_id::text,request_digest FROM privacy_request WHERE tenant_id=$1::uuid AND requested_by=$2::uuid AND idempotency_key=$3 FOR UPDATE`, access.tenantID, access.userID, idempotencyKey).Scan(&publicID, &storedDigest); err != nil { return fmt.Errorf("read idempotent privacy request: %w", err) } if storedDigest != digest { return ErrIdempotencyConflict } return nil } if err != nil { return fmt.Errorf("create privacy request: %w", err) } created = true eventType := "privacy.access_requested" action := "privacy.access.request" if input.RequestType == "ACCESS" { if _, err := tx.Exec(ctx, `INSERT INTO privacy_subtask(privacy_request_id,subsystem,target_type,target_ref) VALUES($1::uuid,'postgresql','subject_data_inventory',$2)`, internalID, subjectPublicID); err != nil { return fmt.Errorf("create privacy subtask: %w", err) } } else { tasks := [][2]string{{"redis", "ephemeral_keys"}, {"export_files", "private_exports"}, {"object_storage", "subject_media"}, {"pgvector", "memory_embeddings"}, {"provider_index", "external_memory_index"}, {"postgresql", "subject_personal_data"}} for _, task := range tasks { if _, err := tx.Exec(ctx, `INSERT INTO privacy_subtask(privacy_request_id,subsystem,target_type,target_ref) VALUES($1::uuid,$2,$3,$4)`, internalID, task[0], task[1], subjectPublicID); err != nil { return fmt.Errorf("create deletion subtask %s: %w", task[0], err) } } if _, err := tx.Exec(ctx, `UPDATE subject_profile SET status='deleting',profile_version=profile_version+1,updated_by=$3::uuid,updated_at=now() WHERE tenant_id=$1::uuid AND id=$2::uuid`, access.tenantID, access.subjectID, access.userID); err != nil { return err } if _, err := tx.Exec(ctx, `UPDATE consent_record SET withdrawn_at=COALESCE(withdrawn_at,GREATEST(now(),granted_at)),scope='{}'::jsonb,evidence_ref=NULL,version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`, access.tenantID, access.subjectID); err != nil { return err } if _, err := tx.Exec(ctx, `UPDATE curated_memory SET status='DISABLED',valid_to=COALESCE(valid_to,now()),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status NOT IN ('DELETED','DISABLED')`, access.tenantID, access.subjectID); err != nil { return err } if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='STALE',last_error='subject privacy deletion requested',updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status<>'DELETED'`, access.tenantID, access.subjectID); err != nil { return err } eventType = "privacy.delete_requested" action = "privacy.delete.request" } dedupe := strings.ReplaceAll(eventType, ".", "-") + ":" + internalID payload, _ := json.Marshal(map[string]any{"request_id": internalID}) if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "privacy_request", AggregateID: internalID, EventType: eventType, DeduplicationKey: &dedupe, Payload: payload}); err != nil { return err } return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: action, ResourceType: "privacy_request", ResourceID: strPtr(publicID), RequestID: requestID, Result: "success", Detail: map[string]any{"subject_id": subjectPublicID, "scope": input.Scope}}) }) if err != nil { return PrivacyRequest{}, false, err } result, err := s.GetPrivacyRequest(ctx, principal, publicID) return result, created, err } func (s *Service) ListPrivacyRequests(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]PrivacyRequest, error) { access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilityPrivacyRead, false) if err != nil { return nil, err } rows, err := s.pool.Query(ctx, privacyRequestSelectSQL+` WHERE pr.tenant_id=$1::uuid AND pr.subject_id=$2::uuid ORDER BY pr.created_at DESC,pr.id DESC`, access.tenantID, access.subjectID) if err != nil { return nil, err } defer rows.Close() items := make([]PrivacyRequest, 0) for rows.Next() { item, err := scanPrivacyRequest(rows) if err != nil { return nil, err } items = append(items, item) } return items, rows.Err() } func (s *Service) GetPrivacyRequest(ctx context.Context, principal identity.Principal, publicID string) (PrivacyRequest, error) { item, err := scanPrivacyRequest(s.pool.QueryRow(ctx, privacyRequestSelectSQL+` JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id AND h.status<>'deleted' JOIN household_member hm ON hm.household_id=h.id AND hm.tenant_id=h.tenant_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=h.tenant_id AND ua.status='active' WHERE pr.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject)) if errors.Is(err, pgx.ErrNoRows) { return PrivacyRequest{}, ErrAccessDenied } if err != nil { return PrivacyRequest{}, err } var role string if err := s.pool.QueryRow(ctx, `SELECT hm.role FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id JOIN household_member hm ON hm.household_id=sp.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=pr.tenant_id AND ua.status='active' WHERE pr.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject).Scan(&role); err != nil { return PrivacyRequest{}, ErrAccessDenied } if !identity.Can(identity.Role(role), identity.CapabilityPrivacyRead) { return PrivacyRequest{}, ErrAccessDenied } if err := s.loadPrivacySubtasks(ctx, &item); err != nil { return PrivacyRequest{}, err } return item, nil } const privacyRequestSelectSQL = `SELECT pr.public_id::text,sp.public_id::text,pr.request_type,pr.scope,pr.status,pr.deadline_at,pr.legal_hold,pr.result_summary,pr.version,pr.created_at,pr.updated_at,pr.completed_at FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id AND sp.tenant_id=pr.tenant_id` func scanPrivacyRequest(row rowScanner) (PrivacyRequest, error) { var item PrivacyRequest var scopeJSON, resultJSON []byte if err := row.Scan(&item.ID, &item.SubjectID, &item.RequestType, &scopeJSON, &item.Status, &item.DeadlineAt, &item.LegalHold, &resultJSON, &item.Version, &item.CreatedAt, &item.UpdatedAt, &item.CompletedAt); err != nil { return PrivacyRequest{}, err } if err := decodeObject(scopeJSON, &item.Scope); err != nil { return PrivacyRequest{}, err } if err := decodeObject(resultJSON, &item.ResultSummary); err != nil { return PrivacyRequest{}, err } item.Subtasks = []PrivacySubtask{} return item, nil } func (s *Service) loadPrivacySubtasks(ctx context.Context, item *PrivacyRequest) error { rows, err := s.pool.Query(ctx, `SELECT ps.public_id::text,ps.subsystem,ps.target_type,ps.target_ref,ps.status,ps.attempts,ps.next_attempt_at,COALESCE(ps.last_error,''),ps.completed_at FROM privacy_subtask ps JOIN privacy_request pr ON pr.id=ps.privacy_request_id WHERE pr.public_id=$1::uuid ORDER BY ps.created_at,ps.id`, item.ID) if err != nil { return err } defer rows.Close() for rows.Next() { var subtask PrivacySubtask if err := rows.Scan(&subtask.ID, &subtask.Subsystem, &subtask.TargetType, &subtask.TargetRef, &subtask.Status, &subtask.Attempts, &subtask.NextAttemptAt, &subtask.LastError, &subtask.CompletedAt); err != nil { return err } item.Subtasks = append(item.Subtasks, subtask) } return rows.Err() }