package outbox

import (
	"context"
	"errors"
	"os"
	"testing"

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

func TestConsumeTransactionRollbackAndIdempotency(t *testing.T) {
	databaseURL := os.Getenv("OUTBOX_TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("OUTBOX_TEST_DATABASE_URL is not configured")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	store := NewStore(pool)
	const subscriber = "outbox-integration-test"
	const eventID = "33333333-3333-4333-8333-333333333333"
	_, _ = pool.Exec(ctx, `DELETE FROM gateway.event_consumptions WHERE subscriber=$1 AND event_id=$2`, subscriber, eventID)
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM gateway.event_consumptions WHERE subscriber=$1 AND event_id=$2`, subscriber, eventID)
	})

	calls := 0
	wantErr := errors.New("rollback handler")
	consumed, err := store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
		calls++
		return wantErr
	})
	if consumed || !errors.Is(err, wantErr) {
		t.Fatalf("failed handler must roll back: consumed=%v err=%v", consumed, err)
	}
	consumed, err = store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
		calls++
		return nil
	})
	if !consumed || err != nil {
		t.Fatalf("second attempt must consume: consumed=%v err=%v", consumed, err)
	}
	consumed, err = store.Consume(ctx, subscriber, eventID, func(context.Context, pgx.Tx) error {
		calls++
		return nil
	})
	if consumed || err != nil || calls != 2 {
		t.Fatalf("duplicate must skip handler: consumed=%v calls=%d err=%v", consumed, calls, err)
	}
}