story/internal/providermemory/processor_integration_test.go

package providermemory

import (
	"context"
	"os"
	"testing"

	"data-terminal-cloud/internal/outbox"

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

type recordingClient struct {
	upserts []Memory
	deletes []string
}

func (c *recordingClient) Upsert(_ context.Context, memory Memory) (string, error) {
	c.upserts = append(c.upserts, memory)
	return "provider-ref-" + memory.MemoryVersion, nil
}
func (c *recordingClient) Delete(_ context.Context, providerRef, _ string) error {
	c.deletes = append(c.deletes, providerRef)
	return nil
}

func TestProcessorSynchronizesAndDeletesProviderMemory(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()
	memoryID, memoryPublicID, versionID, versionPublicID := uuid.NewString(), uuid.NewString(), 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)
	}()
	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,'喜欢在周末照料兰花','[{"type":"integration"}]'::jsonb,'10000000-0000-0000-0000-000000000101','integration',repeat('a',64))`, versionID, versionPublicID, memoryID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=$2::uuid WHERE id=$1::uuid`, memoryID, versionID); 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,status) VALUES('10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,$2::uuid,'provider','PENDING')`, memoryID, versionID); err != nil {
		t.Fatal(err)
	}

	client := &recordingClient{}
	processor, err := NewProcessor(pool, client)
	if err != nil {
		t.Fatal(err)
	}
	payload := []byte(`{"memory_version_id":"` + versionID + `"}`)
	if err := processor.HandleSync(ctx, outbox.Event{AggregateID: memoryID, Payload: payload}); err != nil {
		t.Fatal(err)
	}
	var status string
	var providerRef *string
	if err := pool.QueryRow(ctx, `SELECT status,provider_ref FROM memory_index_entry WHERE memory_version_id=$1::uuid AND index_type='provider'`, versionID).Scan(&status, &providerRef); err != nil {
		t.Fatal(err)
	}
	if status != "ACTIVE" || providerRef == nil || len(client.upserts) != 1 || client.upserts[0].SubjectID != "10000000-0000-0000-0000-000000000411" {
		t.Fatalf("status=%s provider=%v upserts=%+v", status, providerRef, client.upserts)
	}
	if err := processor.HandleDelete(ctx, outbox.Event{AggregateID: memoryID}); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT status,provider_ref FROM memory_index_entry WHERE memory_version_id=$1::uuid AND index_type='provider'`, versionID).Scan(&status, &providerRef); err != nil {
		t.Fatal(err)
	}
	if status != "DELETED" || providerRef != nil || len(client.deletes) != 1 {
		t.Fatalf("status=%s provider=%v deletes=%v", status, providerRef, client.deletes)
	}
}