package feeds

import (
	"context"
	"errors"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/jobs"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestClaimNextPendingSyncJobConcurrentDistinct(t *testing.T) {
	dsn := strings.TrimSpace(os.Getenv("DATABASE_URL"))
	if dsn == "" {
		t.Skip("DATABASE_URL not set")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	pg, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	// Clean up while the pool is still open (t.Cleanup runs after deferred Close).
	defer pg.Close()

	companyID := uuid.New()
	_, err = pg.Exec(ctx, `INSERT INTO companies (id, name) VALUES ($1, $2)`,
		companyID, "sync-claim-"+companyID.String()[:8])
	if err != nil {
		t.Fatalf("seed company: %v", err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM companies WHERE id = $1`, companyID)
	}()

	var feedID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO input_feeds (company_id, name, url, feed_type, status, sync_interval_minutes, options)
		VALUES ($1, 'claim-feed', 'https://example.com/feed.csv', 'csv', 'active', 60, '{}'::jsonb)
		RETURNING id`, companyID).Scan(&feedID)
	if err != nil {
		t.Fatalf("seed feed: %v", err)
	}

	// Live dev workers also ClaimNextPendingSyncJob; seed a buffer so N concurrent
	// claimants still succeed when the worker steals a few pending rows.
	const n = 4
	seedN := n + jobs.MaxSyncWorkers + 4
	jobIDs := make([]uuid.UUID, 0, seedN)
	for i := 0; i < seedN; i++ {
		var id uuid.UUID
		err = pg.QueryRow(ctx, `
			INSERT INTO feed_sync_jobs (feed_id, company_id, status)
			VALUES ($1, $2, 'pending') RETURNING id`, feedID, companyID).Scan(&id)
		if err != nil {
			t.Fatalf("seed sync job: %v", err)
		}
		jobIDs = append(jobIDs, id)
	}
	defer func() {
		for _, id := range jobIDs {
			_, _ = pg.Exec(context.Background(), `DELETE FROM feed_sync_jobs WHERE id = $1`, id)
		}
	}()

	svc := &Service{Pool: pg}
	claimed := make([]uuid.UUID, n)
	var wg sync.WaitGroup
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func(i int) {
			defer wg.Done()
			deadline := time.Now().Add(3 * time.Second)
			for {
				id, _, _, claimErr := svc.ClaimNextPendingSyncJob(ctx)
				if claimErr == nil {
					claimed[i] = id
					return
				}
				if !errors.Is(claimErr, pgx.ErrNoRows) {
					t.Errorf("ClaimNextPendingSyncJob: %v", claimErr)
					return
				}
				if time.Now().After(deadline) {
					t.Errorf("ClaimNextPendingSyncJob: no rows after retries")
					return
				}
				time.Sleep(5 * time.Millisecond)
			}
		}(i)
	}
	wg.Wait()

	seen := make(map[uuid.UUID]struct{}, n)
	for _, id := range claimed {
		if id == uuid.Nil {
			t.Fatal("nil claim")
		}
		if _, ok := seen[id]; ok {
			t.Fatalf("duplicate claim %s (SKIP LOCKED failed)", id)
		}
		seen[id] = struct{}{}
	}
}