story/internal/privacy/delete_processor_integration_test.go

package privacy

import (
	"context"
	"errors"
	"os"
	"testing"
	"time"

	"data-terminal-cloud/internal/cache"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/outbox"

	"github.com/jackc/pgx/v5/pgxpool"
)

const (
	privacyDeleteSubjectID       = "90000000-0000-0000-0000-000000000741"
	privacyDeleteSubjectPublicID = "90000000-0000-0000-0000-000000000742"
	privacyDeleteRequestID       = "90000000-0000-0000-0000-000000000743"
	privacyDeleteRequestPublicID = "90000000-0000-0000-0000-000000000744"
)

type privacyDeleteStore struct {
	removed []string
	failure error
}

func (s *privacyDeleteStore) Remove(_ context.Context, reference string) error {
	if s.failure != nil {
		return s.failure
	}
	s.removed = append(s.removed, reference)
	return nil
}

func TestDeleteProcessorReconcilesAllSubsystemsAndCreatesTombstone(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	redisAddress := os.Getenv("TEST_REDIS_ADDR")
	if databaseURL == "" || redisAddress == "" {
		t.Skip("TEST_DATABASE_URL and TEST_REDIS_ADDR are required")
	}
	ctx := context.Background()
	pool, err := database.Open(ctx, databaseURL, "privacy-delete-test")
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	cleanupPrivacyDeleteFixture(t, ctx, pool)
	defer cleanupPrivacyDeleteFixture(t, context.Background(), pool)

	if _, err := pool.Exec(ctx, `INSERT INTO subject_profile (
		id,public_id,tenant_id,household_id,preferred_name,status
	) VALUES (
		$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',
		'10000000-0000-0000-0000-000000000201','隐私删除集成长者','active'
	)`, privacyDeleteSubjectID, privacyDeleteSubjectPublicID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO privacy_request (
		id,public_id,tenant_id,subject_id,request_type,scope,requested_by,idempotency_key,request_digest
	) VALUES (
		$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',$3::uuid,
		'DELETE','{"reason_code":"subject_request"}'::jsonb,
		'10000000-0000-0000-0000-000000000101','privacy-delete-integration',repeat('9',64)
	)`, privacyDeleteRequestID, privacyDeleteRequestPublicID, privacyDeleteSubjectID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO export_task (
		id,public_id,tenant_id,subject_id,requested_by,scope,status,object_ref,object_size,content_digest,
		expires_at,completed_at,idempotency_key,request_digest
	) VALUES (
		'90000000-0000-0000-0000-000000000745','90000000-0000-0000-0000-000000000746',
		'10000000-0000-0000-0000-000000000001',$1::uuid,'10000000-0000-0000-0000-000000000101',
		'{}'::jsonb,'READY','privacy-delete/export.zip',128,repeat('a',64),now()+interval '1 hour',now(),
		'privacy-delete-export-integration',repeat('b',64)
	)`, privacyDeleteSubjectID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO privacy_subtask (
		public_id,privacy_request_id,subsystem,target_type,target_ref
	) SELECT gen_random_uuid(),$1::uuid,subsystem,'subject_data',subsystem
	FROM unnest(ARRAY['redis','export_files','object_storage','pgvector','provider_index','postgresql','call_memory']) AS subsystem`, privacyDeleteRequestID); err != nil {
		t.Fatal(err)
	}

	redisClient := cache.Open(redisAddress, "", 0)
	defer redisClient.Close()
	store := &privacyDeleteStore{}
	processor, err := NewDeleteProcessor(pool, redisClient, store)
	if err != nil {
		t.Fatal(err)
	}
	store.failure = outbox.RetryAfter(errors.New("provider deletion rate limited"), time.Minute)
	if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || !outbox.IsRetryable(err) || outbox.RetryDelay(err) != time.Minute {
		t.Fatalf("dependency failure lost retry metadata: %v", err)
	}
	var failureStatus, failedSubsystem, taskStatus, lastError string
	var nextAttempt time.Time
	if err := pool.QueryRow(ctx, `SELECT pr.status,pr.result_summary->>'failed_subsystem',ps.status,ps.last_error,ps.next_attempt_at FROM privacy_request pr JOIN privacy_subtask ps ON ps.privacy_request_id=pr.id AND ps.subsystem='export_files' WHERE pr.id=$1::uuid`, privacyDeleteRequestID).Scan(&failureStatus, &failedSubsystem, &taskStatus, &lastError, &nextAttempt); err != nil {
		t.Fatal(err)
	}
	if failureStatus != "PARTIAL_FAILED" || failedSubsystem != "export_files" || taskStatus != "FAILED" || lastError != "provider deletion rate limited" || time.Until(nextAttempt) < 55*time.Second {
		t.Fatalf("failure was hidden or cooldown lost: request=%s subsystem=%s task=%s error=%q next=%v", failureStatus, failedSubsystem, taskStatus, lastError, nextAttempt)
	}
	// A dependency without Retry-After must use the same minimum cooldown in
	// the durable receipt and the error returned to the outbox worker.
	store.failure = errors.New("temporary object removal failure")
	if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || outbox.RetryDelay(err) != 30*time.Second {
		t.Fatalf("ordinary dependency failure did not preserve the receipt cooldown: %v", err)
	}
	var redisAttempts, exportAttempts int
	if err := pool.QueryRow(ctx, `SELECT attempts FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='redis'`, privacyDeleteRequestID).Scan(&redisAttempts); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT attempts,next_attempt_at FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='export_files'`, privacyDeleteRequestID).Scan(&exportAttempts, &nextAttempt); err != nil {
		t.Fatal(err)
	}
	if redisAttempts != 1 || exportAttempts != 2 || time.Until(nextAttempt) < 25*time.Second {
		t.Fatalf("retry repeated completed work or advertised a different cooldown: redis=%d export=%d next=%v", redisAttempts, exportAttempts, nextAttempt)
	}
	store.failure = nil
	if _, err := pool.Exec(ctx, `UPDATE memory_user_state SET legacy_projection_pending=true WHERE subject_id=$1::uuid`, privacyDeleteSubjectID); err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || outbox.RetryDelay(err) != time.Minute {
		t.Fatalf("unfinished call memory cleanup was acknowledged: %v", err)
	}
	if err := pool.QueryRow(ctx, `SELECT pr.status,pr.result_summary->>'failed_subsystem',ps.last_error FROM privacy_request pr JOIN privacy_subtask ps ON ps.privacy_request_id=pr.id AND ps.subsystem='call_memory' WHERE pr.id=$1::uuid`, privacyDeleteRequestID).Scan(&failureStatus, &failedSubsystem, &lastError); err != nil {
		t.Fatal(err)
	}
	if failureStatus != "PARTIAL_FAILED" || failedSubsystem != "call_memory" || lastError != "call memory deletion pending: legacy_projection" {
		t.Fatalf("cleanup recovery reason was hidden: status=%s subsystem=%s reason=%q", failureStatus, failedSubsystem, lastError)
	}
	if _, err := pool.Exec(ctx, `UPDATE memory_user_state SET legacy_projection_pending=false WHERE subject_id=$1::uuid`, privacyDeleteSubjectID); err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err != nil {
		t.Fatal(err)
	}
	var requestStatus, subjectStatus, preferredName string
	var completed, skipped, tombstones, audits int
	if err := pool.QueryRow(ctx, `SELECT status FROM privacy_request WHERE id=$1::uuid`, privacyDeleteRequestID).Scan(&requestStatus); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT status,preferred_name FROM subject_profile WHERE id=$1::uuid`, privacyDeleteSubjectID).Scan(&subjectStatus, &preferredName); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FILTER(WHERE status='COMPLETED'),count(*) FILTER(WHERE status='SKIPPED') FROM privacy_subtask WHERE privacy_request_id=$1::uuid`, privacyDeleteRequestID).Scan(&completed, &skipped); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM tombstone WHERE deletion_request_id=$1::uuid`, privacyDeleteRequestID).Scan(&tombstones); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE action='privacy.delete.complete' AND resource_id=$1::uuid`, privacyDeleteRequestPublicID).Scan(&audits); err != nil {
		t.Fatal(err)
	}
	var exportStatus string
	var exportReference *string
	if err := pool.QueryRow(ctx, `SELECT status,object_ref FROM export_task WHERE id='90000000-0000-0000-0000-000000000745'::uuid`).Scan(&exportStatus, &exportReference); err != nil {
		t.Fatal(err)
	}
	if requestStatus != "COMPLETED" || subjectStatus != "deleted" || preferredName != "[已删除]" || completed != 6 || skipped != 1 || tombstones != 1 || audits != 1 || len(store.removed) != 1 || store.removed[0] != "privacy-delete/export.zip" || exportStatus != "EXPIRED" || exportReference != nil {
		t.Fatalf("request=%s subject=%s name=%q completed=%d skipped=%d tombstones=%d audits=%d removed=%v", requestStatus, subjectStatus, preferredName, completed, skipped, tombstones, audits, store.removed)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT attempts FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='export_files'`, privacyDeleteRequestID).Scan(&exportAttempts); err != nil {
		t.Fatal(err)
	}
	if exportAttempts != 3 || len(store.removed) != 1 {
		t.Fatalf("completed deletion was executed again: attempts=%d removed=%d", exportAttempts, len(store.removed))
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: "90000000-0000-0000-0000-000000000799"}); err != nil {
		t.Fatalf("missing deletion request error=%v", err)
	}
}

func cleanupPrivacyDeleteFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	statements := []string{
		`DELETE FROM tombstone WHERE deletion_request_id='90000000-0000-0000-0000-000000000743'::uuid`,
		`DELETE FROM audit_event WHERE action='privacy.delete.complete' AND resource_id='90000000-0000-0000-0000-000000000744'::uuid`,
		`DELETE FROM privacy_request WHERE id='90000000-0000-0000-0000-000000000743'::uuid`,
		`DELETE FROM export_task WHERE id='90000000-0000-0000-0000-000000000745'::uuid`,
		`DELETE FROM memory_user_state WHERE subject_id='90000000-0000-0000-0000-000000000741'::uuid`,
		`DELETE FROM subject_profile WHERE id='90000000-0000-0000-0000-000000000741'::uuid`,
	}
	for _, statement := range statements {
		if _, err := pool.Exec(ctx, statement); err != nil {
			t.Fatalf("cleanup privacy delete fixture: %v", err)
		}
	}
}