go-trustlog/api/persistence/pulsar_integration_test.go

package persistence

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	_ "github.com/lib/pq"

	"go.yandata.net/iod/iod/go-trustlog/api/adapter"
	"go.yandata.net/iod/iod/go-trustlog/api/logger"
)

const (
	testPulsarURL   = "pulsar://localhost:6650"
	testPulsarTopic = "persistent://public/default/trustlog-integration-test"
)

// setupPulsarPublisher 创建 Pulsar 发布者
func setupPulsarPublisher(t *testing.T) (*adapter.Publisher, bool) {
	log := logger.GetGlobalLogger()
	
	config := adapter.PublisherConfig{
		URL: testPulsarURL,
	}
	
	publisher, err := adapter.NewPublisher(config, log)
	if err != nil {
		t.Logf("Pulsar not available: %v (skipping)", err)
		return nil, false
	}
	
	// 测试连接 - 发送一条测试消息
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	
	testMsg := message.NewMessage(watermill.NewUUID(), []byte("connection-test"))
	err = publisher.Publish(testPulsarTopic, testMsg)
	if err != nil {
		t.Logf("Pulsar connection failed: %v (skipping)", err)
		publisher.Close()
		return nil, false
	}
	
	_ = ctx
	t.Logf("✅ Pulsar connected: %s", testPulsarURL)
	return publisher, true
}

// TestPulsar_Basic 测试基本的 Pulsar 发布
func TestPulsar_Basic(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	publisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	defer publisher.Close()
	
	// 发布测试消息
	testContent := fmt.Sprintf("test-message-%d", time.Now().Unix())
	msg := message.NewMessage(watermill.NewUUID(), []byte(testContent))
	
	err := publisher.Publish(testPulsarTopic, msg)
	if err != nil {
		t.Fatalf("Failed to publish message: %v", err)
	}
	
	t.Logf("✅ Published message: %s (UUID: %s)", testContent, msg.UUID)
	
	// 等待消息发送完成
	time.Sleep(100 * time.Millisecond)
	
	t.Logf("✅ Pulsar basic test passed")
}

// TestPulsar_MultipleMessages 测试批量发布消息
func TestPulsar_MultipleMessages(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	publisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	defer publisher.Close()
	
	// 批量发布多条消息
	messageCount := 10
	messages := make([]*message.Message, messageCount)
	
	for i := 0; i < messageCount; i++ {
		content := fmt.Sprintf("batch-message-%d-%d", time.Now().Unix(), i)
		messages[i] = message.NewMessage(watermill.NewUUID(), []byte(content))
	}
	
	err := publisher.Publish(testPulsarTopic, messages...)
	if err != nil {
		t.Fatalf("Failed to publish messages: %v", err)
	}
	
	t.Logf("✅ Published %d messages", messageCount)
	
	// 等待消息发送完成
	time.Sleep(200 * time.Millisecond)
	
	t.Logf("✅ Pulsar multiple messages test passed")
}

// TestPulsar_WithPostgreSQL 测试 Pulsar + PostgreSQL 集成
func TestPulsar_WithPostgreSQL(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping integration test in short mode")
	}
	
	// 检查 Pulsar
	testPublisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	testPublisher.Close()
	
	// 检查 PostgreSQL
	db, ok := setupPostgresDB(t)
	if !ok {
		t.Skip("PostgreSQL not available")
		return
	}
	defer db.Close()
	
	ctx := context.Background()
	log := logger.GetGlobalLogger()
	
	// 创建 Publisher
	publisherConfig := adapter.PublisherConfig{
		URL: testPulsarURL,
	}
	
	publisher, err := adapter.NewPublisher(publisherConfig, log)
	if err != nil {
		t.Fatalf("Failed to create publisher: %v", err)
	}
	defer publisher.Close()
	
	// 创建 PersistenceManager（仅 DB 策略，用于测试）
	// 注意：DBAndTrustlog 策略需要 Worker 和 Publisher 的完整配置
	// 这里我们测试 DBOnly 策略 + 手动发布到 Pulsar
	config := PersistenceConfig{
		Strategy: StrategyDBOnly,
	}
	
	manager := NewPersistenceManager(db, config, log)
	
	// 保存操作
	op := createTestOperation(t, fmt.Sprintf("pulsar-pg-%d", time.Now().Unix()))
	
	err = manager.SaveOperation(ctx, op)
	if err != nil {
		t.Fatalf("Failed to save operation: %v", err)
	}
	
	t.Logf("✅ Saved operation: %s", op.OpID)
	
	// 验证数据库中的记录
	var count int
	err = db.QueryRowContext(ctx, "SELECT COUNT(*) FROM operation WHERE op_id = $1", op.OpID).Scan(&count)
	if err != nil {
		t.Fatalf("Failed to query database: %v", err)
	}
	
	if count != 1 {
		t.Errorf("Expected 1 record in database, got %d", count)
	}
	
	// 验证状态（DBOnly 策略下，状态应该是 TRUSTLOGGED）
	var status string
	err = db.QueryRowContext(ctx, "SELECT trustlog_status FROM operation WHERE op_id = $1", op.OpID).Scan(&status)
	if err != nil {
		t.Fatalf("Failed to query status: %v", err)
	}
	
	if status != "TRUSTLOGGED" {
		t.Errorf("Expected status TRUSTLOGGED, got %s", status)
	}
	
	t.Logf("✅ Operation saved to database with status: %s", status)
	
	// 手动发布到 Pulsar 来测试完整流程
	msg := message.NewMessage(watermill.NewUUID(), []byte(op.OpID))
	err = publisher.Publish(adapter.OperationTopic, msg)
	if err != nil {
		t.Logf("⚠️  Failed to publish to Pulsar (non-critical for this test): %v", err)
	} else {
		t.Logf("✅ Published operation to Pulsar: %s", op.OpID)
	}
	
	// 等待消息发送
	time.Sleep(200 * time.Millisecond)
	
	t.Logf("✅ Pulsar + PostgreSQL integration test passed")
}

// TestPulsar_HighVolume 测试高并发发布
func TestPulsar_HighVolume(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	publisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	defer publisher.Close()
	
	// 发布100条消息
	messageCount := 100
	messages := make([]*message.Message, messageCount)
	
	for i := 0; i < messageCount; i++ {
		content := fmt.Sprintf("high-volume-test-%d", i)
		messages[i] = message.NewMessage(watermill.NewUUID(), []byte(content))
	}
	
	start := time.Now()
	
	err := publisher.Publish(testPulsarTopic, messages...)
	if err != nil {
		t.Fatalf("Failed to publish messages: %v", err)
	}
	
	duration := time.Since(start)
	
	t.Logf("✅ Published %d messages in %v", messageCount, duration)
	t.Logf("✅ Throughput: %.2f msg/s", float64(messageCount)/duration.Seconds())
	
	// 等待所有消息发送完成
	time.Sleep(500 * time.Millisecond)
	
	t.Logf("✅ Pulsar high volume test passed")
}

// TestPulsar_Reconnect 测试重连机制
func TestPulsar_Reconnect(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	publisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	
	// 发送第一条消息
	msg1 := message.NewMessage(watermill.NewUUID(), []byte("before-close"))
	err := publisher.Publish(testPulsarTopic, msg1)
	if err != nil {
		t.Fatalf("Failed to publish first message: %v", err)
	}
	
	t.Logf("✅ Published first message")
	
	// 关闭并重新创建（模拟重连）
	publisher.Close()
	time.Sleep(100 * time.Millisecond)
	
	publisher, ok = setupPulsarPublisher(t)
	if !ok {
		t.Fatal("Failed to reconnect to Pulsar")
	}
	defer publisher.Close()
	
	// 发送第二条消息
	msg2 := message.NewMessage(watermill.NewUUID(), []byte("after-reconnect"))
	err = publisher.Publish(testPulsarTopic, msg2)
	if err != nil {
		t.Fatalf("Failed to publish after reconnect: %v", err)
	}
	
	t.Logf("✅ Published message after reconnect")
	t.Logf("✅ Pulsar reconnect test passed")
}

// TestPulsar_ErrorHandling 测试错误处理
func TestPulsar_ErrorHandling(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	log := logger.GetGlobalLogger()
	
	// 测试连接到无效的 Pulsar URL
	config := adapter.PublisherConfig{
		URL: "pulsar://invalid-host-that-does-not-exist:9999",
	}
	
	publisher, err := adapter.NewPublisher(config, log)
	if err != nil {
		t.Logf("✅ Expected error for invalid URL: %v", err)
	} else {
		// 如果创建成功，尝试发送消息应该会失败
		msg := message.NewMessage(watermill.NewUUID(), []byte("test"))
		err = publisher.Publish(testPulsarTopic, msg)
		publisher.Close()
		
		if err != nil {
			t.Logf("✅ Expected error when publishing to invalid URL: %v", err)
		} else {
			t.Error("Should have failed to publish to invalid URL")
		}
	}
	
	t.Logf("✅ Pulsar error handling test passed")
}

// TestPulsar_DifferentTopics 测试不同主题
func TestPulsar_DifferentTopics(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping Pulsar integration test in short mode")
	}
	
	publisher, ok := setupPulsarPublisher(t)
	if !ok {
		t.Skip("Pulsar not available")
		return
	}
	defer publisher.Close()
	
	// 发送到不同的主题
	topics := []string{
		"persistent://public/default/test-topic-1",
		"persistent://public/default/test-topic-2",
		"persistent://public/default/test-topic-3",
	}
	
	for _, topic := range topics {
		msg := message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("message-to-%s", topic)))
		err := publisher.Publish(topic, msg)
		if err != nil {
			t.Errorf("Failed to publish to topic %s: %v", topic, err)
		} else {
			t.Logf("✅ Published to topic: %s", topic)
		}
	}
	
	// 等待消息发送完成
	time.Sleep(200 * time.Millisecond)
	
	t.Logf("✅ Pulsar different topics test passed")
}