story/internal/dataexport/processor_integration_test.go

package dataexport

import (
	"context"
	"encoding/json"
	"io"
	"log/slog"
	"os"
	"testing"
	"time"

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

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

const (
	exportTestSubjectID = "90000000-0000-0000-0000-000000000761"
	exportTestTaskID    = "90000000-0000-0000-0000-000000000762"
	exportTestTaskPubID = "90000000-0000-0000-0000-000000000763"
)

type exportTestStore struct {
	reference string
	content   []byte
	digest    string
	removed   []string
}

func (s *exportTestStore) PutJSON(_ context.Context, reference string, content []byte, digest string) (int64, error) {
	s.reference, s.content, s.digest = reference, append([]byte(nil), content...), digest
	return int64(len(content)), nil
}

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

func TestProcessorCreatesAndExpiresScopedExport(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := database.Open(ctx, databaseURL, "data-export-test")
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	cleanupExportFixture(t, ctx, pool)
	defer cleanupExportFixture(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,'90000000-0000-0000-0000-000000000764','10000000-0000-0000-0000-000000000001',
		'10000000-0000-0000-0000-000000000201','数据导出集成长者','active'
	)`, exportTestSubjectID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO export_task (
		id,public_id,tenant_id,subject_id,requested_by,scope,idempotency_key,request_digest
	) VALUES (
		$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',$3::uuid,
		'10000000-0000-0000-0000-000000000101','{"categories":["profile","consents"]}'::jsonb,
		'data-export-integration',repeat('c',64)
	)`, exportTestTaskID, exportTestTaskPubID, exportTestSubjectID); err != nil {
		t.Fatal(err)
	}
	store := &exportTestStore{}
	processor, err := NewProcessor(pool, store, time.Hour)
	if err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: exportTestTaskID}); err != nil {
		t.Fatal(err)
	}
	var status, objectRef, digest string
	var size int64
	if err := pool.QueryRow(ctx, `SELECT status,object_ref,object_size,content_digest FROM export_task WHERE id=$1::uuid`, exportTestTaskID).Scan(&status, &objectRef, &size, &digest); err != nil {
		t.Fatal(err)
	}
	var document struct {
		SchemaVersion string                     `json:"schema_version"`
		Scope         []string                   `json:"scope"`
		Data          map[string]json.RawMessage `json:"data"`
	}
	if err := json.Unmarshal(store.content, &document); err != nil {
		t.Fatal(err)
	}
	if status != "READY" || objectRef != store.reference || size != int64(len(store.content)) || digest != store.digest || document.SchemaVersion != "subject-export-v1" || len(document.Scope) != 2 || len(document.Data) != 2 {
		t.Fatalf("status=%s ref=%s size=%d digest=%s document=%#v", status, objectRef, size, digest, document)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: exportTestTaskID}); err != nil {
		t.Fatal(err)
	}
	if err := processor.Handle(ctx, outbox.Event{AggregateID: "90000000-0000-0000-0000-000000000799"}); err != nil {
		t.Fatalf("missing export error=%v", err)
	}
	if _, err := pool.Exec(ctx, `UPDATE export_task SET expires_at=now()-interval '1 second' WHERE id=$1::uuid`, exportTestTaskID); err != nil {
		t.Fatal(err)
	}
	scanner, err := NewExpiryScanner(pool, store, slog.New(slog.NewTextHandler(io.Discard, nil)), time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	scanner.scan(ctx)
	var reference *string
	if err := pool.QueryRow(ctx, `SELECT status,object_ref FROM export_task WHERE id=$1::uuid`, exportTestTaskID).Scan(&status, &reference); err != nil {
		t.Fatal(err)
	}
	if status != "EXPIRED" || reference != nil || len(store.removed) != 1 || store.removed[0] != store.reference {
		t.Fatalf("expired status=%s reference=%v removed=%v", status, reference, store.removed)
	}
	canceledContext, cancel := context.WithCancel(ctx)
	cancel()
	scanner.Run(canceledContext)
}

func cleanupExportFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	for _, statement := range []string{
		`DELETE FROM audit_event WHERE action IN ('export.ready','export.expire') AND resource_id='90000000-0000-0000-0000-000000000763'::uuid`,
		`DELETE FROM export_task WHERE id='90000000-0000-0000-0000-000000000762'::uuid`,
		`DELETE FROM subject_profile WHERE id='90000000-0000-0000-0000-000000000761'::uuid`,
	} {
		if _, err := pool.Exec(ctx, statement); err != nil {
			t.Fatalf("cleanup export fixture: %v", err)
		}
	}
}