package outbox

import (
	"fmt"
	"net/url"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"gorm.io/driver/postgres"
	"gorm.io/gorm"
)

func TestPostgresConcurrentWorkersDoNotClaimSameMessage(t *testing.T) {
	dsn := os.Getenv("SENSE_OUTBOX_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("set SENSE_OUTBOX_TEST_DATABASE_URL to run the PostgreSQL multi-worker test")
	}
	base, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
	if err != nil {
		t.Fatal(err)
	}
	schema := fmt.Sprintf("sense_outbox_78_%d", time.Now().UnixNano())
	if err = base.Exec("CREATE SCHEMA " + schema).Error; err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = base.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
	scoped, err := gorm.Open(postgres.Open(withSearchPath(dsn, schema)), &gorm.Config{})
	if err != nil {
		t.Fatal(err)
	}
	if err = scoped.AutoMigrate(&Message{}, &DeliveryRecord{}, &Attempt{}); err != nil {
		t.Fatal(err)
	}
	now := time.Date(2026, 8, 28, 12, 0, 0, 0, time.UTC)
	for index := 0; index < 20; index++ {
		if _, err = Enqueue(scoped, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: fmt.Sprintf("event-%d", index), IdempotencyKey: fmt.Sprintf("event:%d:v1", index), PayloadJSON: []byte(`{}`)}, now); err != nil {
			t.Fatal(err)
		}
	}
	workers := []string{"worker-a", "worker-b"}
	results := make(chan []Message, len(workers))
	errorsCh := make(chan error, len(workers))
	var group sync.WaitGroup
	for _, worker := range workers {
		group.Add(1)
		go func(name string) {
			defer group.Done()
			relay := NewRelay(scoped)
			relay.Now = func() time.Time { return now }
			items, claimErr := relay.Claim(name, 20)
			if claimErr != nil {
				errorsCh <- claimErr
				return
			}
			results <- items
		}(worker)
	}
	group.Wait()
	close(results)
	close(errorsCh)
	for claimErr := range errorsCh {
		t.Fatal(claimErr)
	}
	seen := map[string]string{}
	for batch := range results {
		for _, item := range batch {
			if owner, exists := seen[item.ID]; exists {
				t.Fatalf("message %s claimed by %s and %s", item.ID, owner, item.LeaseOwner)
			}
			seen[item.ID] = item.LeaseOwner
		}
	}
	if len(seen) != 20 {
		t.Fatalf("claimed=%d want=20", len(seen))
	}
}

func withSearchPath(dsn, schema string) string {
	if strings.Contains(dsn, "://") {
		parsed, err := url.Parse(dsn)
		if err == nil {
			query := parsed.Query()
			query.Set("search_path", schema)
			parsed.RawQuery = query.Encode()
			return parsed.String()
		}
	}
	return strings.TrimSpace(dsn) + " search_path=" + schema
}