package persistence

import (
	"context"
	"database/sql/driver"
	"errors"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// TestClaimEvent_AtomicClaim covers the Round 2 C1 fix: ClaimEvent must
// issue a single atomic UPDATE guarded on Status='pending' and interpret
// RowsAffected via the ErrOutboxEventAlreadyClaimed sentinel. The prior
// SELECT ... FOR UPDATE SKIP LOCKED form silently degraded to a
// non-locking read under TiDB optimistic txn mode, so concurrent workers
// could both "claim" the same row.
//
// Round 3 extends the predicate to also reclaim stale 'processing' rows
// (see TestClaimEvent_ReclaimsStaleProcessing / TestClaimEvent_SkipsActiveProcessing);
// the pending-only cases below still pass because the added OR-branch
// is orthogonal.
func TestClaimEvent_AtomicClaim(t *testing.T) {
	t.Run("claim succeeds when row is pending", func(t *testing.T) {
		db, mock, err := sqlmock.New()
		require.NoError(t, err)
		defer db.Close()

		repo := NewOutboxRepository(db)

		// The builder emits UPDATE ... WHERE Id=? AND (Status=? OR (Status=? AND LastModifiedDate < ?))
		// with the params in that order. We match on the UPDATE keyword
		// and the presence of the pending-guard param; the column list
		// ordering inside SET is map-iteration-order and not worth
		// pinning down. The final AnyArg covers the stale-cutoff time.
		// Observed SQL (see race-test output):
		//   UPDATE ... SET Status=?, LastModifiedDate=?
		//   WHERE (`Status` = ? OR (`Status` = ? AND `LastModifiedDate` < ?))
		//   AND   `_System_OutboxEvent`.`Id` = ?
		// so the param order is:
		//   SET args(2) || raw-where args(3) || Id(1)
		mock.ExpectExec("(?i)^UPDATE ").
			WithArgs(
				sqlmock.AnyArg(), // SET Status='processing'
				sqlmock.AnyArg(), // SET LastModifiedDate=time.Now()
				"pending",        // WHERE pending branch
				"processing",     // WHERE stale-processing branch
				sqlmock.AnyArg(), // staleCutoff time.Time
				"evt-1",          // AND Id=?
			).
			WillReturnResult(sqlmock.NewResult(0, 1))

		id, err := repo.ClaimEvent(context.Background(), "evt-1")
		require.NoError(t, err)
		assert.Equal(t, "evt-1", id)
		require.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("claim returns sentinel when row is no longer pending", func(t *testing.T) {
		db, mock, err := sqlmock.New()
		require.NoError(t, err)
		defer db.Close()

		repo := NewOutboxRepository(db)

		mock.ExpectExec("(?i)^UPDATE ").
			WithArgs(
				sqlmock.AnyArg(),
				sqlmock.AnyArg(),
				"pending",
				"processing",
				sqlmock.AnyArg(),
				"evt-2",
			).
			WillReturnResult(sqlmock.NewResult(0, 0)) // RowsAffected == 0

		id, err := repo.ClaimEvent(context.Background(), "evt-2")
		assert.Empty(t, id)
		require.Error(t, err)
		assert.True(t, errors.Is(err, ErrOutboxEventAlreadyClaimed),
			"expected ErrOutboxEventAlreadyClaimed, got %v", err)
		require.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("claim surfaces DB errors", func(t *testing.T) {
		db, mock, err := sqlmock.New()
		require.NoError(t, err)
		defer db.Close()

		repo := NewOutboxRepository(db)

		mock.ExpectExec("(?i)^UPDATE ").
			WillReturnError(errors.New("connection reset"))

		id, err := repo.ClaimEvent(context.Background(), "evt-3")
		assert.Empty(t, id)
		require.Error(t, err)
		assert.False(t, errors.Is(err, ErrOutboxEventAlreadyClaimed),
			"raw DB errors must not be confused with the already-claimed sentinel")
		require.NoError(t, mock.ExpectationsWereMet())
	})
}

// TestClaimEvent_ReclaimsStaleProcessing covers the Round 3 C2 fix: when a
// prior worker died between ClaimEvent and UpdateStatus, the row sits in
// 'processing' with a LastModifiedDate older than OutboxStaleProcessingSeconds.
// A new ClaimEvent must win that row via the OR-branch of the WHERE
// predicate so the event eventually dispatches instead of being stranded.
//
// Sqlmock cannot actually inspect the row state; what we verify here is
// that the emitted UPDATE carries both the pending-guard AND the
// stale-processing guard with a cutoff that is strictly in the past (and
// within a reasonable window of OutboxStaleProcessingSeconds). The real
// TiDB behavior (only rows matching the OR-clause are updated) is covered
// by the local-runtime-verify lane on new-prod-essential.
func TestClaimEvent_ReclaimsStaleProcessing(t *testing.T) {
	db, mock, err := sqlmock.New()
	require.NoError(t, err)
	defer db.Close()

	repo := NewOutboxRepository(db)

	mock.ExpectExec("(?i)^UPDATE ").
		WithArgs(
			sqlmock.AnyArg(),     // SET Status='processing'
			sqlmock.AnyArg(),     // SET LastModifiedDate=time.Now()
			"pending",            // WHERE pending branch
			"processing",         // WHERE stale-processing branch
			staleCutoffMatcher{}, // cutoff must be ~OutboxStaleProcessingSeconds ago
			"evt-stale",          // AND Id=?
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	id, err := repo.ClaimEvent(context.Background(), "evt-stale")
	require.NoError(t, err)
	assert.Equal(t, "evt-stale", id)
	require.NoError(t, mock.ExpectationsWereMet())
}

// TestClaimEvent_SkipsActiveProcessing covers the other side of the Round
// 3 C2 fix: if the row is in 'processing' but LastModifiedDate is fresh
// (the prior worker is alive and still dispatching), the reclaim predicate
// does NOT match and RowsAffected==0 — we must surface the
// ErrOutboxEventAlreadyClaimed sentinel so the service layer treats the
// attempt as a benign skip, not a failure.
func TestClaimEvent_SkipsActiveProcessing(t *testing.T) {
	db, mock, err := sqlmock.New()
	require.NoError(t, err)
	defer db.Close()

	repo := NewOutboxRepository(db)

	mock.ExpectExec("(?i)^UPDATE ").
		WithArgs(
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			"pending",
			"processing",
			sqlmock.AnyArg(),
			"evt-active",
		).
		WillReturnResult(sqlmock.NewResult(0, 0)) // no rows matched

	id, err := repo.ClaimEvent(context.Background(), "evt-active")
	assert.Empty(t, id)
	require.Error(t, err)
	assert.True(t, errors.Is(err, ErrOutboxEventAlreadyClaimed),
		"fresh 'processing' rows must surface ErrOutboxEventAlreadyClaimed")
	require.NoError(t, mock.ExpectationsWereMet())
}

// staleCutoffMatcher accepts any time.Time that sits roughly
// OutboxStaleProcessingSeconds before now (±5s slack for test scheduling
// jitter). This guards the reclaim-window configuration so a future typo
// on OutboxStaleProcessingSeconds or the cutoff computation is caught
// without tying the test to wall-clock precision.
type staleCutoffMatcher struct{}

func (staleCutoffMatcher) Match(v driver.Value) bool {
	t, ok := v.(time.Time)
	if !ok {
		return false
	}
	delta := time.Since(t)
	window := time.Duration(OutboxStaleProcessingSeconds) * time.Second
	return delta >= window-5*time.Second && delta <= window+5*time.Second
}