story/internal/memorylifecycle/delete_processor_integration_test.go

package memorylifecycle

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"os"
	"strings"
	"testing"

	"data-terminal-cloud/internal/developmentembedding"
	"data-terminal-cloud/internal/outbox"

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

func TestDeleteProcessorRedactsMemoryAndIsIdempotent(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if _, err := NewDeleteProcessor(nil); err == nil {
		t.Fatal("nil database pool was accepted")
	}
	processor, err := NewDeleteProcessor(pool)
	if err != nil {
		t.Fatal(err)
	}
	memoryID, memoryPublicID := uuid.NewString(), uuid.NewString()
	versionID, versionPublicID := uuid.NewString(), uuid.NewString()
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE resource_id=$1::uuid`, memoryPublicID)
		_, _ = pool.Exec(ctx, `DELETE FROM memory_index_entry WHERE memory_id=$1::uuid`, memoryID)
		_, _ = pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=NULL WHERE id=$1::uuid`, memoryID)
		_, _ = pool.Exec(ctx, `DELETE FROM memory_version WHERE memory_id=$1::uuid`, memoryID)
		_, _ = pool.Exec(ctx, `DELETE FROM curated_memory WHERE id=$1::uuid`, memoryID)
	}()
	content := "这段合成记忆必须被可靠删除。"
	if _, err := pool.Exec(ctx, `
		INSERT INTO curated_memory(id,public_id,tenant_id,subject_id,status,sensitivity,valid_from)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401','ACTIVE','normal',now())`, memoryID, memoryPublicID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO memory_version(id,public_id,memory_id,version_no,content,source_refs,confirmed_by,change_reason,content_digest)
		VALUES($1::uuid,$2::uuid,$3::uuid,1,$4,'[{"type":"integration"}]','10000000-0000-0000-0000-000000000101','原始原因',$5)`,
		versionID, versionPublicID, memoryID, content, strings.Repeat("c", 64)); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=$1::uuid WHERE id=$2::uuid`, versionID, memoryID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,embedding,provider_ref,status,indexed_at)
		VALUES('10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,$2::uuid,'pgvector',$3::vector,'integration','ACTIVE',now())`,
		memoryID, versionID, developmentembedding.Literal(developmentembedding.Vector(content))); err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: uuid.NewString()}); err != nil {
		t.Fatalf("missing memory deletion: %v", err)
	}
	event := outbox.Event{AggregateID: memoryID}
	if err := processor.Handle(ctx, event); err != nil {
		t.Fatalf("active memory no-op: %v", err)
	}
	var status string
	if err := pool.QueryRow(ctx, `SELECT status FROM curated_memory WHERE id=$1::uuid`, memoryID).Scan(&status); err != nil || status != "ACTIVE" {
		t.Fatalf("active memory status=%s err=%v", status, err)
	}
	if _, err := pool.Exec(ctx, `UPDATE curated_memory SET status='PENDING_DELETE',valid_to=now() WHERE id=$1::uuid`, memoryID); err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, event); err != nil {
		t.Fatalf("delete memory: %v", err)
	}
	var deletedAtSet, embeddingIsNull, providerIsNull bool
	var version int64
	if err := pool.QueryRow(ctx, `SELECT status,version,deleted_at IS NOT NULL FROM curated_memory WHERE id=$1::uuid`, memoryID).Scan(&status, &version, &deletedAtSet); err != nil {
		t.Fatal(err)
	}
	if status != "DELETED" || version != 2 || !deletedAtSet {
		t.Fatalf("deleted memory status=%s version=%d deleted_at=%v", status, version, deletedAtSet)
	}
	var redactedContent, changeReason, contentDigest, indexStatus string
	if err := pool.QueryRow(ctx, `SELECT content,change_reason,content_digest FROM memory_version WHERE id=$1::uuid`, versionID).Scan(&redactedContent, &changeReason, &contentDigest); err != nil {
		t.Fatal(err)
	}
	expectedDigest := sha256.Sum256([]byte(deletedMemoryMarker))
	if redactedContent != deletedMemoryMarker || !strings.Contains(changeReason, "user requested deletion") || contentDigest != hex.EncodeToString(expectedDigest[:]) {
		t.Fatalf("redacted content=%q reason=%q digest=%q", redactedContent, changeReason, contentDigest)
	}
	if err := pool.QueryRow(ctx, `SELECT status,embedding IS NULL,provider_ref IS NULL FROM memory_index_entry WHERE memory_id=$1::uuid`, memoryID).Scan(&indexStatus, &embeddingIsNull, &providerIsNull); err != nil {
		t.Fatal(err)
	}
	if indexStatus != "DELETED" || !embeddingIsNull || !providerIsNull {
		t.Fatalf("deleted index status=%s embedding_null=%v provider_null=%v", indexStatus, embeddingIsNull, providerIsNull)
	}
	if err := processor.Handle(ctx, event); err != nil {
		t.Fatalf("replay deletion: %v", err)
	}
	var auditCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE resource_id=$1::uuid AND action='memory.delete.complete'`, memoryPublicID).Scan(&auditCount); err != nil || auditCount != 1 {
		t.Fatalf("deletion audit count=%d err=%v", auditCount, err)
	}
}