story/internal/privacy/delete_processor.go

package privacy

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/outbox"
	"data-terminal-cloud/internal/providermemory"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"
)

type deletionObjectStore interface {
	Remove(context.Context, string) error
}

type DeleteProcessor struct {
	pool           *pgxpool.Pool
	redis          *redis.Client
	store          deletionObjectStore
	audit          *audit.Repository
	providerMemory providermemory.Client
}

type deletionRequest struct {
	id, tenantID, subjectID, subjectPublicID, publicID, status, requestType string
}

type deletionSubtask struct {
	id, subsystem, status string
}

func NewDeleteProcessor(pool *pgxpool.Pool, redisClient *redis.Client, store deletionObjectStore, providerClients ...providermemory.Client) (*DeleteProcessor, error) {
	if pool == nil || redisClient == nil || store == nil {
		return nil, errors.New("database, redis and object store are required")
	}
	if len(providerClients) > 1 {
		return nil, errors.New("at most one provider memory client is allowed")
	}
	var providerClient providermemory.Client
	if len(providerClients) == 1 {
		providerClient = providerClients[0]
	}
	return &DeleteProcessor{pool: pool, redis: redisClient, store: store, audit: audit.NewRepository(), providerMemory: providerClient}, nil
}

func (p *DeleteProcessor) Handle(ctx context.Context, event outbox.Event) error {
	request, err := p.loadRequest(ctx, event.AggregateID)
	if err != nil || request.id == "" || request.status == "COMPLETED" {
		return err
	}
	if request.requestType != "DELETE" {
		return fmt.Errorf("privacy request type %s is not handled by delete processor", request.requestType)
	}
	if _, err := p.pool.Exec(ctx, `UPDATE privacy_request SET status='VERIFYING',version=version+1,updated_at=now() WHERE id=$1::uuid AND status='SUBMITTED'`, request.id); err != nil {
		return err
	}
	if _, err := p.pool.Exec(ctx, `UPDATE privacy_request SET status='EXECUTING',completed_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid AND status IN ('VERIFYING','PARTIAL_FAILED')`, request.id); err != nil {
		return err
	}

	tasks, err := p.loadSubtasks(ctx, request.id)
	if err != nil {
		return err
	}
	for _, task := range tasks {
		if task.status == "COMPLETED" || task.status == "SKIPPED" {
			continue
		}
		if _, err := p.pool.Exec(ctx, `UPDATE privacy_subtask SET status='RUNNING',attempts=attempts+1,next_attempt_at=NULL,last_error=NULL,completed_at=NULL,updated_at=now() WHERE id=$1::uuid`, task.id); err != nil {
			return err
		}
		outcome, note, runErr := p.executeSubtask(ctx, request, task.subsystem)
		if runErr != nil {
			message := safeDeletionError(runErr)
			_, _ = p.pool.Exec(ctx, `UPDATE privacy_subtask SET status='FAILED',last_error=$2,next_attempt_at=now()+interval '30 seconds',updated_at=now() WHERE id=$1::uuid`, task.id, message)
			_, _ = p.pool.Exec(ctx, `UPDATE privacy_request SET status='PARTIAL_FAILED',result_summary=jsonb_build_object('failed_subsystem',$2,'content_included',false),version=version+1,updated_at=now() WHERE id=$1::uuid`, request.id, task.subsystem)
			return fmt.Errorf("delete subject subsystem %s: %w", task.subsystem, runErr)
		}
		if _, err := p.pool.Exec(ctx, `UPDATE privacy_subtask SET status=$2,last_error=NULLIF($3,''),next_attempt_at=NULL,completed_at=now(),updated_at=now() WHERE id=$1::uuid`, task.id, outcome, note); err != nil {
			return err
		}
	}

	return database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
		var open int
		if err := tx.QueryRow(ctx, `SELECT count(*) FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND status NOT IN ('COMPLETED','SKIPPED')`, request.id).Scan(&open); err != nil {
			return err
		}
		if open != 0 {
			return errors.New("privacy deletion reconciliation still has open subtasks")
		}
		result, _ := json.Marshal(map[string]any{"subtasks_reconciled": len(tasks), "content_included": false, "completed_at": time.Now().UTC()})
		if _, err := tx.Exec(ctx, `UPDATE privacy_request SET status='COMPLETED',result_summary=$2::jsonb,completed_at=now(),version=version+1,updated_at=now() WHERE id=$1::uuid`, request.id, result); err != nil {
			return err
		}
		digest := sha256.Sum256([]byte(request.subjectPublicID))
		if _, err := tx.Exec(ctx, `INSERT INTO tombstone(tenant_id,resource_type,resource_digest,deletion_request_id,retention_reason) VALUES($1::uuid,'subject',$2,$3::uuid,'privacy deletion proof; contains no subject content') ON CONFLICT (tenant_id,resource_type,resource_digest) DO NOTHING`, request.tenantID, hex.EncodeToString(digest[:]), request.id); err != nil {
			return err
		}
		return p.audit.Record(ctx, tx, audit.Event{TenantID: &request.tenantID, ActorType: "system", Action: "privacy.delete.complete", ResourceType: "privacy_request", ResourceID: &request.publicID, Result: "success", Detail: map[string]any{"subject_digest": hex.EncodeToString(digest[:]), "subtasks_reconciled": len(tasks), "content_included": false}})
	})
}

func (p *DeleteProcessor) loadRequest(ctx context.Context, id string) (deletionRequest, error) {
	var value deletionRequest
	err := p.pool.QueryRow(ctx, `SELECT pr.id::text,pr.tenant_id::text,pr.subject_id::text,sp.public_id::text,pr.public_id::text,pr.status,pr.request_type FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id WHERE pr.id=$1::uuid`, id).Scan(&value.id, &value.tenantID, &value.subjectID, &value.subjectPublicID, &value.publicID, &value.status, &value.requestType)
	if errors.Is(err, pgx.ErrNoRows) {
		return deletionRequest{}, nil
	}
	return value, err
}

func (p *DeleteProcessor) loadSubtasks(ctx context.Context, requestID string) ([]deletionSubtask, error) {
	rows, err := p.pool.Query(ctx, `SELECT id::text,subsystem,status FROM privacy_subtask WHERE privacy_request_id=$1::uuid ORDER BY CASE subsystem WHEN 'redis' THEN 1 WHEN 'export_files' THEN 2 WHEN 'object_storage' THEN 3 WHEN 'pgvector' THEN 4 WHEN 'provider_index' THEN 5 WHEN 'postgresql' THEN 6 ELSE 99 END,id`, requestID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []deletionSubtask{}
	for rows.Next() {
		var item deletionSubtask
		if err := rows.Scan(&item.id, &item.subsystem, &item.status); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (p *DeleteProcessor) executeSubtask(ctx context.Context, request deletionRequest, subsystem string) (string, string, error) {
	switch subsystem {
	case "redis":
		keys, err := p.subjectRedisKeys(ctx, request.tenantID, request.subjectID)
		if err != nil {
			return "", "", err
		}
		if len(keys) > 0 {
			if err := p.redis.Del(ctx, keys...).Err(); err != nil {
				return "", "", err
			}
		}
		return "COMPLETED", "", nil
	case "export_files":
		rows, err := p.pool.Query(ctx, `SELECT object_ref FROM export_task WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND object_ref IS NOT NULL`, request.tenantID, request.subjectID)
		if err != nil {
			return "", "", err
		}
		refs := []string{}
		for rows.Next() {
			var ref string
			if err := rows.Scan(&ref); err != nil {
				rows.Close()
				return "", "", err
			}
			refs = append(refs, ref)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return "", "", err
		}
		for _, ref := range refs {
			if err := p.store.Remove(ctx, ref); err != nil {
				return "", "", outbox.Retryable(err)
			}
		}
		_, err = p.pool.Exec(ctx, `UPDATE export_task SET status='EXPIRED',scope='{}'::jsonb,object_ref=NULL,object_size=NULL,content_digest=NULL,expires_at=NULL,last_error=NULL,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`, request.tenantID, request.subjectID)
		return "COMPLETED", "", err
	case "object_storage":
		rows, err := p.pool.Query(ctx, `SELECT ms.object_ref FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id WHERE c.tenant_id=$1::uuid AND c.subject_id=$2::uuid AND ms.object_ref IS NOT NULL`, request.tenantID, request.subjectID)
		if err != nil {
			return "", "", err
		}
		refs := []string{}
		for rows.Next() {
			var ref string
			if err := rows.Scan(&ref); err != nil {
				rows.Close()
				return "", "", err
			}
			refs = append(refs, ref)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return "", "", err
		}
		for _, ref := range refs {
			if err := p.store.Remove(ctx, ref); err != nil {
				return "", "", outbox.Retryable(err)
			}
		}
		_, err = p.pool.Exec(ctx, `UPDATE media_segment ms SET status='DELETED',object_ref=NULL,expected_digest=repeat('0',64),expected_size=1,duration_ms=1,last_error_code=NULL,deleted_at=COALESCE(deleted_at,now()),updated_at=now() FROM conversation c WHERE ms.conversation_id=c.id AND c.tenant_id=$1::uuid AND c.subject_id=$2::uuid`, request.tenantID, request.subjectID)
		return "COMPLETED", "", err
	case "pgvector":
		_, err := p.pool.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,provider_ref=NULL,status='DELETED',last_error=NULL,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND index_type='pgvector'`, request.tenantID, request.subjectID)
		return "COMPLETED", "", err
	case "provider_index":
		rows, err := p.pool.Query(ctx, `SELECT memory_version_id::text,provider_ref FROM memory_index_entry WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND index_type='provider' AND provider_ref IS NOT NULL AND status<>'DELETED' ORDER BY created_at,id`, request.tenantID, request.subjectID)
		if err != nil {
			return "", "", err
		}
		type providerRef struct{ versionID, value string }
		refs := []providerRef{}
		for rows.Next() {
			var ref providerRef
			if err := rows.Scan(&ref.versionID, &ref.value); err != nil {
				rows.Close()
				return "", "", err
			}
			refs = append(refs, ref)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return "", "", err
		}
		if len(refs) > 0 && p.providerMemory == nil {
			return "", "", errors.New("external provider index references exist but no deletion adapter is configured")
		}
		for _, ref := range refs {
			if err := p.providerMemory.Delete(ctx, ref.value, "privacy-provider-memory-delete:"+request.id+":"+ref.versionID); err != nil {
				return "", "", outbox.Retryable(err)
			}
		}
		_, err = p.pool.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,provider_ref=NULL,status='DELETED',last_error=NULL,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND index_type='provider'`, request.tenantID, request.subjectID)
		if len(refs) == 0 {
			return "SKIPPED", "no external provider index references were present", err
		}
		return "COMPLETED", "", err
	case "postgresql":
		return "COMPLETED", "", p.redactPostgreSQL(ctx, request)
	default:
		return "", "", fmt.Errorf("unsupported deletion subsystem %q", subsystem)
	}
}

func (p *DeleteProcessor) subjectRedisKeys(ctx context.Context, tenantID, subjectID string) ([]string, error) {
	rows, err := p.pool.Query(ctx, `SELECT DISTINCT c.public_id::text,d.public_id::text FROM conversation c JOIN device d ON d.id=c.device_id AND d.tenant_id=c.tenant_id WHERE c.tenant_id=$1::uuid AND c.subject_id=$2::uuid`, tenantID, subjectID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	keys := []string{}
	for rows.Next() {
		var conversationID, deviceID string
		if err := rows.Scan(&conversationID, &deviceID); err != nil {
			return nil, err
		}
		keys = append(keys, "voice-platform:conversation:lease:"+conversationID, "voice-platform:device:presence:"+deviceID)
	}
	return keys, rows.Err()
}

func (p *DeleteProcessor) redactPostgreSQL(ctx context.Context, request deletionRequest) error {
	return database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
		statements := []string{
			`UPDATE subject_fact SET content='[已按隐私请求删除]',source_ref=NULL,confirmation_status='superseded',valid_to=COALESCE(valid_to,now()),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE conversation_turn SET final_text='[已按隐私请求删除]',provider_ref=NULL WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE conversation_summary SET content='[已按隐私请求删除]',source_turn_ids='["privacy-redacted"]'::jsonb,status='SUPERSEDED',last_error=NULL WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE conversation_content_assessment SET decision_reason='[privacy-redacted]',model_ref='privacy-redacted',prompt_version='privacy-redacted' WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE story_revision sr SET content='[已按隐私请求删除]',edit_reason=NULL,model_ref=NULL,prompt_version=NULL FROM story_draft sd WHERE sr.story_id=sd.id AND sd.tenant_id=$1::uuid AND sd.subject_id=$2::uuid`,
			`UPDATE story_draft SET title='[已按隐私请求删除]',status='SUPERSEDED',review_reason=NULL,version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE fact_candidate SET content='[已按隐私请求删除]',time_range='{}'::jsonb,source_turn_ids='["privacy-redacted"]'::jsonb,status='SUPERSEDED',review_reason=NULL,version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE fact_candidate_revision fcr SET content='[已按隐私请求删除]',edit_reason=NULL FROM fact_candidate fc WHERE fcr.fact_candidate_id=fc.id AND fc.tenant_id=$1::uuid AND fc.subject_id=$2::uuid`,
			`UPDATE memory_version mv SET content='[已按隐私请求删除]',source_refs='[{"type":"privacy_tombstone"}]'::jsonb,change_reason=NULL,content_digest=encode(digest('[已按隐私请求删除]','sha256'),'hex') FROM curated_memory cm WHERE mv.memory_id=cm.id AND cm.tenant_id=$1::uuid AND cm.subject_id=$2::uuid`,
			`UPDATE curated_memory SET status='DELETED',valid_to=COALESCE(valid_to,now()),deleted_at=COALESCE(deleted_at,now()),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE memory_index_entry SET embedding=NULL,provider_ref=NULL,status='DELETED',last_error=NULL,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`DELETE FROM conversation_memory_use WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`DELETE FROM memory_tool_call WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE content_review_signal SET reason='[privacy-redacted]' WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE consent_snapshot SET effective_consents='{}'::jsonb,source_consent_ids='[]'::jsonb WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE context_snapshot SET memory_versions='[]'::jsonb,content_digest=repeat('0',64),sanitized_summary=NULL WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE recording_access_grant SET purpose='[privacy-redacted]',revoked_at=COALESCE(revoked_at,now()),updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE sharing_grant SET purpose='[privacy-redacted]',revoked_at=COALESCE(revoked_at,now()),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE sharing_access_log sal SET reason_code=CASE WHEN result='DENIED' THEN 'PRIVACY_REDACTED' ELSE NULL END FROM sharing_grant sg WHERE sal.sharing_grant_id=sg.id AND sg.tenant_id=$1::uuid AND sg.subject_id=$2::uuid`,
			`UPDATE media_access_log mal SET purpose='[privacy-redacted]',reason_code=NULL FROM conversation c WHERE mal.conversation_id=c.id AND c.tenant_id=$1::uuid AND c.subject_id=$2::uuid`,
			`UPDATE upload_batch ub SET manifest_digest=repeat('0',64),last_error_code=NULL,status=CASE WHEN ub.status='COMPLETED' THEN ub.status ELSE 'EXPIRED' END,updated_at=now() FROM conversation c WHERE ub.conversation_id=c.id AND c.tenant_id=$1::uuid AND c.subject_id=$2::uuid`,
			`UPDATE provider_event pe SET sanitized_payload='{}'::jsonb,last_error=NULL FROM conversation c WHERE pe.conversation_id=c.id AND c.tenant_id=$1::uuid AND c.subject_id=$2::uuid`,
			`UPDATE device_binding SET unbound_at=COALESCE(unbound_at,now()),unbound_by=COALESCE(unbound_by,(SELECT requested_by FROM privacy_request WHERE id=$3::uuid)),reason='subject privacy deletion' WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
			`UPDATE subject_profile SET preferred_name='[已删除]',profile_data='{}'::jsonb,conversation_preferences='{}'::jsonb,status='deleted',deleted_at=COALESCE(deleted_at,now()),profile_version=profile_version+1,updated_at=now() WHERE tenant_id=$1::uuid AND id=$2::uuid`,
			`DELETE FROM web_event WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
		}
		for _, statement := range statements {
			args := []any{request.tenantID, request.subjectID}
			if strings.Contains(statement, "$3") {
				args = append(args, request.id)
			}
			if _, err := tx.Exec(ctx, statement, args...); err != nil {
				return err
			}
		}
		if _, err := tx.Exec(ctx, `UPDATE outbox_event SET payload='{"privacy_redacted":true}'::jsonb,status=CASE WHEN status IN ('pending','processing') THEN 'completed' ELSE status END,completed_at=CASE WHEN status IN ('pending','processing') THEN COALESCE(completed_at,now()) ELSE completed_at END,locked_at=NULL,locked_by=NULL,updated_at=now() WHERE tenant_id=$1::uuid AND aggregate_id IN (SELECT id FROM conversation WHERE subject_id=$2::uuid UNION SELECT id FROM curated_memory WHERE subject_id=$2::uuid UNION SELECT id FROM export_task WHERE subject_id=$2::uuid UNION SELECT ub.public_id FROM upload_batch ub JOIN conversation c ON c.id=ub.conversation_id WHERE c.subject_id=$2::uuid)`, request.tenantID, request.subjectID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `UPDATE audit_event SET purpose=NULL,detail='{"privacy_redacted":true}'::jsonb WHERE tenant_id=$1::uuid AND (detail->>'subject_id'=(SELECT public_id::text FROM subject_profile WHERE id=$2::uuid) OR resource_id IN (SELECT public_id FROM subject_profile WHERE id=$2::uuid UNION SELECT public_id FROM subject_fact WHERE subject_id=$2::uuid UNION SELECT public_id FROM consent_record WHERE subject_id=$2::uuid UNION SELECT public_id FROM conversation WHERE subject_id=$2::uuid UNION SELECT public_id FROM story_draft WHERE subject_id=$2::uuid UNION SELECT public_id FROM curated_memory WHERE subject_id=$2::uuid UNION SELECT public_id FROM export_task WHERE subject_id=$2::uuid UNION SELECT public_id FROM privacy_request WHERE subject_id=$2::uuid UNION SELECT public_id FROM recording_access_grant WHERE subject_id=$2::uuid UNION SELECT public_id FROM sharing_grant WHERE subject_id=$2::uuid UNION SELECT ub.public_id FROM upload_batch ub JOIN conversation c ON c.id=ub.conversation_id WHERE c.subject_id=$2::uuid UNION SELECT ms.public_id FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id WHERE c.subject_id=$2::uuid))`, request.tenantID, request.subjectID); err != nil {
			return err
		}
		return nil
	})
}

func safeDeletionError(err error) string {
	if err == nil {
		return ""
	}
	value := err.Error()
	if len(value) > 450 {
		return value[:450]
	}
	return value
}