package idempotency

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

func TestExecutorReplaysAndRejectsHashConflict(t *testing.T) {
	pool := integrationPool(t)
	executor := NewExecutor(pool)
	key := fmt.Sprintf("test-%d", time.Now().UnixNano())
	tenantID := "10000000-0000-0000-0000-000000000001"
	request := Request{
		TenantID: &tenantID, ActorKey: "integration-user", Scope: "integration.create",
		Key: key, RequestHash: Hash([]byte(`{"value":1}`)), ExpiresAfter: time.Hour,
	}
	var calls atomic.Int32
	operation := func(context.Context, pgx.Tx) (Result, error) {
		calls.Add(1)
		return Result{StatusCode: 201, Body: json.RawMessage(`{"created":true}`)}, nil
	}
	first, err := executor.Execute(context.Background(), request, operation)
	if err != nil || first.Replayed {
		t.Fatalf("first execute = %#v, %v", first, err)
	}
	second, err := executor.Execute(context.Background(), request, operation)
	if err != nil || !second.Replayed || string(second.Body) != `{"created": true}` {
		t.Fatalf("second execute = %#v, %v", second, err)
	}
	if calls.Load() != 1 {
		t.Fatalf("operation calls = %d, want 1", calls.Load())
	}
	lookedUp, found, err := executor.Lookup(context.Background(), request)
	if err != nil || !found || !lookedUp.Replayed || string(lookedUp.Body) != `{"created": true}` {
		t.Fatalf("lookup=%#v found=%v err=%v", lookedUp, found, err)
	}
	request.RequestHash = Hash([]byte(`{"value":2}`))
	if _, err := executor.Execute(context.Background(), request, operation); !errors.Is(err, ErrKeyConflict) {
		t.Fatalf("hash conflict error = %v, want ErrKeyConflict", err)
	}
}

func TestExecutorSerializesConcurrentRequests(t *testing.T) {
	pool := integrationPool(t)
	executor := NewExecutor(pool)
	key := fmt.Sprintf("concurrent-%d", time.Now().UnixNano())
	request := Request{
		ActorKey: "integration-concurrent", Scope: "integration.create", Key: key,
		RequestHash: Hash([]byte(`{"value":1}`)), ExpiresAfter: time.Hour,
	}
	var calls atomic.Int32
	operation := func(context.Context, pgx.Tx) (Result, error) {
		calls.Add(1)
		time.Sleep(30 * time.Millisecond)
		return Result{StatusCode: 201, Body: json.RawMessage(`{"ok":true}`)}, nil
	}
	var waitGroup sync.WaitGroup
	errorsChannel := make(chan error, 5)
	for range 5 {
		waitGroup.Add(1)
		go func() {
			defer waitGroup.Done()
			_, err := executor.Execute(context.Background(), request, operation)
			errorsChannel <- err
		}()
	}
	waitGroup.Wait()
	close(errorsChannel)
	for err := range errorsChannel {
		if err != nil {
			t.Errorf("concurrent execute returned %v", err)
		}
	}
	if calls.Load() != 1 {
		t.Fatalf("operation calls = %d, want 1", calls.Load())
	}
}

func integrationPool(t *testing.T) *pgxpool.Pool {
	t.Helper()
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	pool, err := pgxpool.New(context.Background(), databaseURL)
	if err != nil {
		t.Fatalf("connect integration database: %v", err)
	}
	t.Cleanup(pool.Close)
	return pool
}