package connector_events

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

	"github.com/DATA-DOG/go-sqlmock"
)

// ─────────────────────────────────────────────────────────────────────────────
// PeekBatch tests
// ─────────────────────────────────────────────────────────────────────────────

// TestPeekBatch_ReturnsRowsInOrderHoldsNoLease verifies that PeekBatch returns
// up to n events in (CreatedDate, DeliverySequence, Id) order using the same
// claimable WHERE predicate as ClaimNext, and issues NO UPDATE (no lease minted).
func TestPeekBatch_ReturnsRowsInOrderHoldsNoLease(t *testing.T) {
	svc, mock := newMock(t) // no idgen tokens needed — peek issues no UPDATE

	// SELECT returns 3 full rows.
	mock.ExpectQuery(`ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC`).
		WillReturnRows(sqlmock.NewRows(allEventColumns()).
			AddRow(eventRowValues("evt-p1", "ns", "pack-1", "webhook.event", "ik1", "pending", 0, "")...).
			AddRow(eventRowValues("evt-p2", "ns", "pack-1", "webhook.event", "ik2", "pending", 0, "")...).
			AddRow(eventRowValues("evt-p3", "ns", "pack-1", "webhook.event", "ik3", "failed", 0, "")...))

	result, err := svc.PeekBatch(context.Background(), "pack-1", 3)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(result) != 3 {
		t.Fatalf("want 3 events, got %d", len(result))
	}
	for i, wantID := range []string{"evt-p1", "evt-p2", "evt-p3"} {
		if result[i].ID != wantID {
			t.Errorf("result[%d].ID = %q, want %q", i, result[i].ID, wantID)
		}
	}
	// No UPDATE must have been issued (no lease claimed).
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations (unexpected UPDATE?): %v", err)
	}
}

func TestPeekBatchForClassesFiltersQueueClass(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`COALESCE\(e\.QueueClass, 'default'\) IN \(\?\)`).
		WithArgs("pack-1", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "bulk", 3).
		WillReturnRows(sqlmock.NewRows(allEventColumns()).
			AddRow(eventRowValues("evt-bulk", "ns", "pack-1", "webhook.event", "ik-bulk", "pending", 0, "")...))

	result, err := svc.PeekBatchForClasses(context.Background(), "pack-1", []string{" bulk "}, 3)
	if err != nil {
		t.Fatalf("PeekBatchForClasses: %v", err)
	}
	if len(result) != 1 || result[0].ID != "evt-bulk" {
		t.Fatalf("result = %v, want evt-bulk", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestPeekBatch_EmptyQueueReturnsNil verifies that an empty queue returns nil
// without error.
func TestPeekBatch_EmptyQueueReturnsNil(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC`).
		WillReturnRows(sqlmock.NewRows(allEventColumns()))

	result, err := svc.PeekBatch(context.Background(), "pack-e", 10)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(result) != 0 {
		t.Fatalf("want empty peek, got %v", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestPeekBatch_StaleProcessingPredicateIncluded verifies that PeekBatch
// uses the same stale-processing predicate as ClaimNext.
func TestPeekBatch_StaleProcessingPredicateIncluded(t *testing.T) {
	svc, mock := newMock(t)

	// The query must include the stale-processing predicate.
	mock.ExpectQuery(`e.Status = 'processing' AND e.LastModifiedDate`).
		WillReturnRows(sqlmock.NewRows(allEventColumns()).
			AddRow(eventRowValues("evt-stale-p", "ns", "pack-s", "webhook.event", "ik-sp", "processing", 1, "")...))

	result, err := svc.PeekBatch(context.Background(), "pack-s", 5)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(result) != 1 || result[0].ID != "evt-stale-p" {
		t.Fatalf("want [evt-stale-p], got %v", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestPeekBatch_DefaultNReturnsOne verifies that n<=0 defaults to 1.
func TestPeekBatch_DefaultNReturnsOne(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`LIMIT`).
		WillReturnRows(sqlmock.NewRows(allEventColumns()).
			AddRow(eventRowValues("evt-def", "ns", "pack-d", "webhook.event", "ik-def", "pending", 0, "")...))

	result, err := svc.PeekBatch(context.Background(), "pack-d", 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(result) != 1 || result[0].ID != "evt-def" {
		t.Fatalf("want [evt-def], got %v", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// ClaimByID tests
// ─────────────────────────────────────────────────────────────────────────────

// TestClaimByID_WinsCAS verifies that when the CAS UPDATE matches 1 row
// (RowsAffected==1), ClaimByID returns a non-empty token and ok=true.
// Shape 1: NO loadByID SELECT is issued — the caller holds the full row
// from PeekBatch and stamps the returned token onto a local copy.
func TestClaimByID_WinsCAS(t *testing.T) {
	svc, mock := newMock(t, "fresh-token-1")

	// CAS UPDATE — wins (RowsAffected=1). NO loadByID SELECT should follow.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).WillReturnResult(sqlmock.NewResult(0, 1))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-claim-1", "worker-c", "pack-1", sql.NullString{}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !ok {
		t.Fatal("want ok=true on CAS win")
	}
	if token == "" {
		t.Fatal("want non-empty token on CAS win")
	}
	if token != "fresh-token-1" {
		t.Fatalf("want token %q, got %q", "fresh-token-1", token)
	}
	// No loadByID SELECT must have been issued.
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations (unexpected SELECT?): %v", err)
	}
}

// TestClaimByID_LosesCAS verifies that RowsAffected==0 returns ("", false, nil)
// — the row was already claimed by another worker.
func TestClaimByID_LosesCAS(t *testing.T) {
	svc, mock := newMock(t, "unused-token")

	// CAS UPDATE — loses (RowsAffected=0). No loadByID SELECT should follow.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).WillReturnResult(sqlmock.NewResult(0, 0))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-lost", "worker-l", "pack-1", sql.NullString{}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if ok {
		t.Fatal("want ok=false on CAS loss")
	}
	if token != "" {
		t.Fatalf("want empty token on CAS loss, got %q", token)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestClaimByID_StalePredicate verifies that ClaimByID uses the stale-processing
// predicate in its WHERE clause (same as ClaimNext). Shape 1: CAS-only, no SELECT.
func TestClaimByID_StalePredicate(t *testing.T) {
	svc, mock := newMock(t, "token-stale-cbi")

	// The UPDATE must include the stale-processing predicate. No SELECT follows.
	mock.ExpectExec(`Status = 'processing' AND .*LastModifiedDate`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-stale-cbi", "worker-s", "pack-1", sql.NullString{}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !ok {
		t.Fatal("want ok=true")
	}
	if token == "" {
		t.Fatal("want non-empty token on CAS win")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations (unexpected SELECT?): %v", err)
	}
}

// TestClaimByID_NoCASLoadOnWin verifies that ClaimByID (shape 1) does NOT
// issue a loadByID SELECT after winning the CAS. The worker holds the full row
// from PeekBatch; only the token needs stamping. An unexpected SELECT would fail
// ExpectationsWereMet here.
func TestClaimByID_NoCASLoadOnWin(t *testing.T) {
	svc, mock := newMock(t, "token-noload-1")

	// Only the UPDATE is expected — no SELECT.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-noload-1", "worker-nl", "pack-1", sql.NullString{}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !ok || token == "" {
		t.Fatalf("want (non-empty token, true); got token=%q ok=%v", token, ok)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unexpected DB call (loadByID SELECT should not exist in shape 1): %v", err)
	}
}

func TestClaimBatchByID_AllRowsWinWithoutWinnerReload(t *testing.T) {
	svc, mock := newMock(t, "batch-token-a", "batch-token-b")

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent.*CASE e\.Id.*NOT EXISTS`).
		WillReturnResult(sqlmock.NewResult(0, 2))

	claimed, err := svc.ClaimBatchByID(context.Background(), []BatchClaimRequest{
		{EventID: "evt-batch-a", WorkerID: "worker-a"},
		{EventID: "evt-batch-b", WorkerID: "worker-b"},
	})
	if err != nil {
		t.Fatalf("ClaimBatchByID: %v", err)
	}
	if len(claimed) != 2 {
		t.Fatalf("claimed len = %d, want 2", len(claimed))
	}
	if got := claimed["evt-batch-a"]; got.WorkerID != "worker-a" || got.ClaimToken != "batch-token-a" {
		t.Fatalf("claim evt-batch-a = %+v, want worker-a/batch-token-a", got)
	}
	if got := claimed["evt-batch-b"]; got.WorkerID != "worker-b" || got.ClaimToken != "batch-token-b" {
		t.Fatalf("claim evt-batch-b = %+v, want worker-b/batch-token-b", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unexpected DB call (full batch win should not reload winners): %v", err)
	}
}

func TestClaimBatchByID_PartialWinLoadsOnlyClaimedRows(t *testing.T) {
	svc, mock := newMock(t, "batch-token-a", "batch-token-b")
	claimedAt := time.Date(2026, 7, 5, 10, 0, 0, 0, time.UTC)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent.*NOT EXISTS`).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT Id, ClaimedBy, ClaimToken, ClaimedAt`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ClaimedBy", "ClaimToken", "ClaimedAt"}).
			AddRow("evt-batch-a", "worker-a", "batch-token-a", claimedAt))

	claimed, err := svc.ClaimBatchByID(context.Background(), []BatchClaimRequest{
		{EventID: "evt-batch-a", WorkerID: "worker-a"},
		{EventID: "evt-batch-b", WorkerID: "worker-b"},
	})
	if err != nil {
		t.Fatalf("ClaimBatchByID: %v", err)
	}
	if len(claimed) != 1 {
		t.Fatalf("claimed len = %d, want 1", len(claimed))
	}
	if got := claimed["evt-batch-a"]; got.WorkerID != "worker-a" || got.ClaimToken != "batch-token-a" || !got.ClaimedAt.Equal(claimedAt) {
		t.Fatalf("claim evt-batch-a = %+v, want worker-a/batch-token-a/%s", got, claimedAt)
	}
	if _, ok := claimed["evt-batch-b"]; ok {
		t.Fatalf("evt-batch-b unexpectedly claimed: %+v", claimed["evt-batch-b"])
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Guard-presence unit tests (sqlmock)
// ─────────────────────────────────────────────────────────────────────────────

// TestPeekBatch_DoesNotApplyOrderingGuard asserts that PeekBatch can observe
// later siblings; ClaimByID remains the ordering CAS arbiter.
func TestPeekBatch_DoesNotApplyOrderingGuard(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT e\.Id, e\.Namespace`).
		WillReturnRows(sqlmock.NewRows(allEventColumns()))

	_, err := svc.PeekBatch(context.Background(), "pack-1", 1)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("PeekBatch SELECT expectation not met: %v", err)
	}
}

// TestClaimByID_NotExistsGuardPresent asserts that ClaimByID's UPDATE contains
// the NOT EXISTS ordering guard — mirroring ClaimNext's UPDATE-side guard.
func TestClaimByID_NotExistsGuardPresent(t *testing.T) {
	svc, mock := newMock(t, "token-guard")

	mock.ExpectExec(`NOT EXISTS`).WillReturnResult(sqlmock.NewResult(0, 1))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-guard", "worker-g",
		"pack-1", sql.NullString{String: "dlv-1", Valid: true}, 1)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !ok || token == "" {
		t.Fatalf("want CAS win; got ok=%v token=%q", ok, token)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("ClaimByID UPDATE missing NOT EXISTS guard: %v", err)
	}
}

// TestClaimByID_OrderingGuardBlocksLaterSibling asserts that ClaimByID returns
// ok=false (guard fired) when RowsAffected==0 — simulating the DB rejecting the
// CAS because a non-terminal earlier sibling exists for the same DeliveryId.
func TestClaimByID_OrderingGuardBlocksLaterSibling(t *testing.T) {
	svc, mock := newMock(t, "token-blocked")

	// RowsAffected=0 simulates the NOT EXISTS guard blocking seq=1 while seq=0
	// is still non-terminal.
	mock.ExpectExec(`NOT EXISTS`).WillReturnResult(sqlmock.NewResult(0, 0))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-seq1", "worker-g",
		"pack-1", sql.NullString{String: "dlv-order", Valid: true}, 1)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if ok {
		t.Fatal("want ok=false — ordering guard must block later sibling when earlier is non-terminal")
	}
	if token != "" {
		t.Fatalf("want empty token on guard block, got %q", token)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestClaimByID_NullDeliveryIdNotBlocked asserts that a NULL-DeliveryId row
// (legacy singleton) is never blocked by the ordering guard — NULL=NULL is false
// in SQL so the subquery never returns a row. Simulated by RowsAffected=1.
func TestClaimByID_NullDeliveryIdNotBlocked(t *testing.T) {
	svc, mock := newMock(t, "token-null-dlv")

	// RowsAffected=1: DB let the claim through (no sibling matched the NULL join).
	mock.ExpectExec(`NOT EXISTS`).WillReturnResult(sqlmock.NewResult(0, 1))

	token, _, ok, err := svc.ClaimByID(context.Background(), "evt-legacy", "worker-g",
		"pack-1", sql.NullString{Valid: false}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !ok || token == "" {
		t.Fatalf("want ok=true for NULL-DeliveryId singleton; got ok=%v token=%q", ok, token)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Integration test: PeekBatch+ClaimByID per-delivery ordering (live TiDB)
// ─────────────────────────────────────────────────────────────────────────────

// TestBatchClaim_StrictPerDeliveryOrder_Integration verifies that the production
// drain path (PeekBatch + ClaimByID) enforces per-delivery execution ordering.
// This test mirrors concurrent_claim_integration_test.go's harness and TiDB
// skip/gating pattern, and runs in CI where TiDB is provisioned (port 4001).
//
// Scenario:
//  1. Two events seq=0 and seq=1 share the same DeliveryId.
//  2. ClaimByID(seq=1) is rejected (ok=false) while seq=0 is non-terminal.
//  3. After marking seq=0 processed, ClaimByID(seq=1) succeeds.
//  4. A NULL-DeliveryId singleton claims freely regardless of other deliveries.
func TestBatchClaim_StrictPerDeliveryOrder_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "batch-order-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "batch-order-test")

	svc := NewService(db, nil, nil)
	ctx := context.Background()
	const packInstall = "install-token-batch-order-test"
	const dlvID = "dlv-batch-order-001"

	// Enqueue seq=0 and seq=1 under the same DeliveryId.
	id0, _, err := svc.Enqueue(ctx, "batch-order-test", "webhook.event", "ikey-seq0", nil, packInstall, 1, "", "", 0, dlvID)
	if err != nil {
		t.Fatalf("Enqueue seq=0: %v", err)
	}
	id1, _, err := svc.Enqueue(ctx, "batch-order-test", "webhook.event", "ikey-seq1", nil, packInstall, 1, "", "", 1, dlvID)
	if err != nil {
		t.Fatalf("Enqueue seq=1: %v", err)
	}

	// Enqueue a NULL-DeliveryId singleton (pass empty deliveryID = legacy).
	idNull, _, err := svc.Enqueue(ctx, "batch-order-test", "webhook.event", "ikey-null", nil, packInstall, 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue null-dlv: %v", err)
	}

	// PeekBatch is observational: it sees seq=0, seq=1, and the NULL singleton.
	// ClaimByID below remains the ordering CAS arbiter.
	peeked, err := svc.PeekBatch(ctx, packInstall, 10)
	if err != nil {
		t.Fatalf("PeekBatch: %v", err)
	}
	peekedIDs := make(map[string]bool)
	for _, ev := range peeked {
		peekedIDs[ev.ID] = true
	}
	if !peekedIDs[id0] {
		t.Errorf("PeekBatch must return seq=0 (%s); got IDs %v", id0, peekedIDs)
	}
	if !peekedIDs[idNull] {
		t.Errorf("PeekBatch must return NULL-dlv singleton (%s); got IDs %v", idNull, peekedIDs)
	}
	if !peekedIDs[id1] {
		t.Errorf("PeekBatch must return seq=1 (%s) so a batch worker can try grouped claim; got IDs %v", id1, peekedIDs)
	}

	// Claim seq=0 via ClaimByID — should win.
	token0, _, ok0, err := svc.ClaimByID(ctx, id0, "worker-b", packInstall,
		sql.NullString{String: dlvID, Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID seq=0: %v", err)
	}
	if !ok0 || token0 == "" {
		t.Fatalf("ClaimByID seq=0 must win; got ok=%v token=%q", ok0, token0)
	}

	// Another worker must still be blocked while seq=0 is in processing.
	_, _, okOther, err := svc.ClaimByID(ctx, id1, "worker-other", packInstall,
		sql.NullString{String: dlvID, Valid: true}, 1)
	if err != nil {
		t.Fatalf("ClaimByID seq=1 from another worker: %v", err)
	}
	if okOther {
		t.Error("ClaimByID seq=1 must block another worker while seq=0 is non-terminal")
	}

	// Reusing the same WorkerID is not proof of owning the earlier lease: a
	// supervisor rebuild may reuse WorkerID after leaving seq=0 in processing.
	_, _, okSame, err := svc.ClaimByID(ctx, id1, "worker-b", packInstall,
		sql.NullString{String: dlvID, Valid: true}, 1)
	if err != nil {
		t.Fatalf("ClaimByID seq=1 from same worker: %v", err)
	}
	if okSame {
		t.Error("ClaimByID seq=1 must block same WorkerID while seq=0 is non-terminal")
	}

	if err := svc.MarkProcessed(ctx, id0, token0); err != nil {
		t.Fatalf("MarkProcessed seq=0: %v", err)
	}
	token1, _, ok1, err := svc.ClaimByID(ctx, id1, "worker-b", packInstall,
		sql.NullString{String: dlvID, Valid: true}, 1)
	if err != nil {
		t.Fatalf("ClaimByID seq=1 after seq=0 processed: %v", err)
	}
	if !ok1 || token1 == "" {
		t.Fatalf("ClaimByID seq=1 must win after seq=0 terminal-success; got ok=%v token=%q", ok1, token1)
	}
	if err := svc.MarkProcessed(ctx, id1, token1); err != nil {
		t.Fatalf("MarkProcessed seq=1: %v", err)
	}

	// Verify NULL-DeliveryId singleton can be claimed freely (independent singleton).
	tokenNull, _, okNull, err := svc.ClaimByID(ctx, idNull, "worker-b", packInstall,
		sql.NullString{Valid: false}, 0)
	if err != nil {
		t.Fatalf("ClaimByID null-dlv: %v", err)
	}
	if !okNull || tokenNull == "" {
		t.Fatalf("ClaimByID null-dlv singleton must win freely; got ok=%v token=%q", okNull, tokenNull)
	}
}

func TestBatchClaim_StaleSameWorkerPredecessorBlocksLaterDeliverySibling_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "batch-stale-worker-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "batch-stale-worker-test")

	ctx := context.Background()
	now := time.Date(2026, 6, 21, 19, 55, 0, 0, time.UTC)
	svc := NewService(db, nil, func() time.Time { return now })
	const packInstall = "install-token-batch-stale-worker-test"
	const dlvID = "dlv-batch-stale-worker-001"

	id0, _, err := svc.Enqueue(ctx, "batch-stale-worker-test", "webhook.event", "ikey-stale-seq0", nil, packInstall, 1, "", "", 0, dlvID)
	if err != nil {
		t.Fatalf("Enqueue stale seq=0: %v", err)
	}
	id1, _, err := svc.Enqueue(ctx, "batch-stale-worker-test", "webhook.event", "ikey-stale-seq1", nil, packInstall, 1, "", "", 1, dlvID)
	if err != nil {
		t.Fatalf("Enqueue stale seq=1: %v", err)
	}

	token0, _, ok0, err := svc.ClaimByID(ctx, id0, "worker-reused", packInstall, sql.NullString{String: dlvID, Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID stale seq=0: %v", err)
	}
	if !ok0 || token0 == "" {
		t.Fatalf("ClaimByID stale seq=0 must win; got ok=%v token=%q", ok0, token0)
	}
	staleAt := now.Add(-(StaleProcessingSeconds + 1) * time.Second)
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET LastModifiedDate = ?
 WHERE Id = ?`, staleAt, id0); err != nil {
		t.Fatalf("force stale predecessor: %v", err)
	}

	_, _, ok1, err := svc.ClaimByID(ctx, id1, "worker-reused", packInstall, sql.NullString{String: dlvID, Valid: true}, 1)
	if err != nil {
		t.Fatalf("ClaimByID stale seq=1 with reused worker: %v", err)
	}
	if ok1 {
		t.Fatal("ClaimByID seq=1 must block behind stale same-worker predecessor until predecessor is reclaimed")
	}

	reclaimedToken0, _, okReclaimed, err := svc.ClaimByID(ctx, id0, "worker-reused", packInstall, sql.NullString{String: dlvID, Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID reclaim stale seq=0: %v", err)
	}
	if !okReclaimed || reclaimedToken0 == "" {
		t.Fatalf("ClaimByID stale seq=0 reclaim must win; got ok=%v token=%q", okReclaimed, reclaimedToken0)
	}
	if err := svc.MarkProcessed(ctx, id0, reclaimedToken0); err != nil {
		t.Fatalf("MarkProcessed reclaimed seq=0: %v", err)
	}
	token1, _, ok1, err := svc.ClaimByID(ctx, id1, "worker-reused", packInstall, sql.NullString{String: dlvID, Valid: true}, 1)
	if err != nil {
		t.Fatalf("ClaimByID seq=1 after stale predecessor processed: %v", err)
	}
	if !ok1 || token1 == "" {
		t.Fatalf("ClaimByID seq=1 must win after predecessor processed; got ok=%v token=%q", ok1, token1)
	}
}

// TestBatchClaim_SerializesSameSourceAcrossDeliveries_Integration verifies that
// the production drain path (PeekBatch + ClaimByID) blocks a later same-source
// row even when it arrived under a different DeliveryId.
func TestBatchClaim_SerializesSameSourceAcrossDeliveries_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "batch-source-order-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "batch-source-order-test")

	baseNow := time.Date(2026, 6, 12, 16, 5, 0, 123000000, time.UTC)
	nowSeq := []time.Time{
		baseNow,
		baseNow.Add(200 * time.Microsecond),
		baseNow.Add(400 * time.Microsecond),
	}
	nowIdx := 0
	svc := NewService(db, deterministicID(
		"evt-batch-source-z-head",
		"evt-batch-source-a-tail",
		"evt-batch-source-m-other",
	), func() time.Time {
		if nowIdx >= len(nowSeq) {
			return nowSeq[len(nowSeq)-1]
		}
		v := nowSeq[nowIdx]
		nowIdx++
		return v
	})
	ctx := context.Background()
	const packInstall = "install-token-batch-source-order-test"

	payloadA1 := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Sent"}}`)
	payloadA2 := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Responded"}}`)
	payloadB1 := []byte(`{"objectKind":"workflow-actions","sourceID":"other-source","payload":{"status":"Sent"}}`)

	idA1, _, err := svc.Enqueue(ctx, "batch-source-order-test", "webhook.event", "ikey-batch-source-a-1", payloadA1, packInstall, 1, "", "", 0, "dlv-batch-source-a-1")
	if err != nil {
		t.Fatalf("Enqueue shared-source first: %v", err)
	}
	idA2, _, err := svc.Enqueue(ctx, "batch-source-order-test", "webhook.event", "ikey-batch-source-a-2", payloadA2, packInstall, 1, "", "", 0, "dlv-batch-source-a-2")
	if err != nil {
		t.Fatalf("Enqueue shared-source second: %v", err)
	}
	idB1, _, err := svc.Enqueue(ctx, "batch-source-order-test", "webhook.event", "ikey-batch-source-b-1", payloadB1, packInstall, 1, "", "", 0, "dlv-batch-source-b-1")
	if err != nil {
		t.Fatalf("Enqueue other-source: %v", err)
	}

	peeked, err := svc.PeekBatch(ctx, packInstall, 10)
	if err != nil {
		t.Fatalf("PeekBatch initial: %v", err)
	}
	peekedIDs := make(map[string]bool)
	for _, ev := range peeked {
		peekedIDs[ev.ID] = true
	}
	if !peekedIDs[idA1] {
		t.Fatalf("PeekBatch must include first shared-source row %s; got %v", idA1, peekedIDs)
	}
	if !peekedIDs[idB1] {
		t.Fatalf("PeekBatch must include other-source row %s; got %v", idB1, peekedIDs)
	}
	if !peekedIDs[idA2] {
		t.Fatalf("PeekBatch must include later shared-source row %s; ClaimByID enforces ordering", idA2)
	}

	tokenA1, _, okA1, err := svc.ClaimByID(ctx, idA1, "worker-batch-source-a", packInstall, sql.NullString{String: "dlv-batch-source-a-1", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID first shared-source row: %v", err)
	}
	if !okA1 || tokenA1 == "" {
		t.Fatalf("ClaimByID first shared-source row must win; got ok=%v token=%q", okA1, tokenA1)
	}

	tokenB1, _, okB1, err := svc.ClaimByID(ctx, idB1, "worker-batch-source-b", packInstall, sql.NullString{String: "dlv-batch-source-b-1", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID other-source row: %v", err)
	}
	if !okB1 || tokenB1 == "" {
		t.Fatalf("ClaimByID other-source row must still be claimable; got ok=%v token=%q", okB1, tokenB1)
	}

	if _, _, ok, err := svc.ClaimByID(ctx, idA2, "worker-batch-source-a2", packInstall, sql.NullString{String: "dlv-batch-source-a-2", Valid: true}, 0); err != nil {
		t.Fatalf("ClaimByID blocked shared-source sibling: %v", err)
	} else if ok {
		t.Fatalf("later shared-source row %s must stay blocked while first is processing", idA2)
	}

	if err := svc.MarkProcessed(ctx, idA1, tokenA1); err != nil {
		t.Fatalf("MarkProcessed first shared-source row: %v", err)
	}

	tokenA2, _, okA2, err := svc.ClaimByID(ctx, idA2, "worker-batch-source-a2", packInstall, sql.NullString{String: "dlv-batch-source-a-2", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID released shared-source sibling: %v", err)
	}
	if !okA2 || tokenA2 == "" {
		t.Fatalf("later shared-source row must become claimable after first is processed; got ok=%v token=%q", okA2, tokenA2)
	}
}

func TestPeekClaimableBatchSkipsBlockedSourceLane_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "batch-claimable-source-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "batch-claimable-source-test")

	baseNow := time.Date(2026, 7, 6, 10, 40, 0, 0, time.UTC)
	nowSeq := []time.Time{
		baseNow,
		baseNow.Add(1 * time.Second),
		baseNow.Add(2 * time.Second),
		baseNow.Add(3 * time.Second),
	}
	nowIdx := 0
	svc := NewService(db, deterministicID(
		"evt-claimable-shared-head",
		"evt-claimable-shared-tail",
		"evt-claimable-other-head",
		"claim-token-shared-head",
	), func() time.Time {
		if nowIdx >= len(nowSeq) {
			return nowSeq[len(nowSeq)-1]
		}
		v := nowSeq[nowIdx]
		nowIdx++
		return v
	})
	ctx := context.Background()
	const packInstall = "install-token-batch-claimable-source-test"

	sharedHeadPayload := []byte(`{"objectKind":"contacts","sourceID":"shared-contact","payload":{"id":"shared-contact"}}`)
	sharedTailPayload := []byte(`{"objectKind":"contacts","sourceID":"shared-contact","payload":{"id":"shared-contact"}}`)
	otherPayload := []byte(`{"objectKind":"contacts","sourceID":"other-contact","payload":{"id":"other-contact"}}`)

	idSharedHead, _, err := svc.Enqueue(ctx, "batch-claimable-source-test", "webhook.event", "ikey-claimable-shared-head", sharedHeadPayload, packInstall, 1, "", "", 0, "dlv-claimable-shared-head")
	if err != nil {
		t.Fatalf("Enqueue shared head: %v", err)
	}
	idSharedTail, _, err := svc.Enqueue(ctx, "batch-claimable-source-test", "webhook.event", "ikey-claimable-shared-tail", sharedTailPayload, packInstall, 1, "", "", 0, "dlv-claimable-shared-tail")
	if err != nil {
		t.Fatalf("Enqueue shared tail: %v", err)
	}
	idOther, _, err := svc.Enqueue(ctx, "batch-claimable-source-test", "webhook.event", "ikey-claimable-other-head", otherPayload, packInstall, 1, "", "", 0, "dlv-claimable-other-head")
	if err != nil {
		t.Fatalf("Enqueue other head: %v", err)
	}

	tokenSharedHead, _, ok, err := svc.ClaimByID(ctx, idSharedHead, "worker-shared-head", packInstall, sql.NullString{String: "dlv-claimable-shared-head", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID shared head: %v", err)
	}
	if !ok || tokenSharedHead == "" {
		t.Fatalf("ClaimByID shared head must win; got ok=%v token=%q", ok, tokenSharedHead)
	}

	rawPeek, err := svc.PeekBatch(ctx, packInstall, 1)
	if err != nil {
		t.Fatalf("PeekBatch raw: %v", err)
	}
	if len(rawPeek) != 1 || rawPeek[0].ID != idSharedTail {
		t.Fatalf("raw PeekBatch should surface blocked shared tail %s first, got %+v", idSharedTail, rawPeek)
	}

	claimablePeek, err := svc.PeekClaimableBatch(ctx, packInstall, 1)
	if err != nil {
		t.Fatalf("PeekClaimableBatch: %v", err)
	}
	if len(claimablePeek) != 1 || claimablePeek[0].ID != idOther {
		t.Fatalf("PeekClaimableBatch should skip blocked shared tail and return %s, got %+v", idOther, claimablePeek)
	}
}

// TestBatchClaim_SerializesSameSourceAcrossEqualEnqueueOrdinals_Integration
// reproduces the workflow-action duplicate lane where two same-source rows
// carry the same EnqueueOrdinal. The tie-break must still pick exactly one
// winner and hold the sibling until the winner is terminal.
func TestBatchClaim_SerializesSameSourceAcrossEqualEnqueueOrdinals_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "batch-source-equal-ordinal-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "batch-source-equal-ordinal-test")

	svc := NewService(db, deterministicID(
		"evt-batch-source-a-head",
		"evt-batch-source-b-tail",
		"claim-token-a",
		"claim-token-b",
	), nil)
	ctx := context.Background()
	const packInstall = "install-token-batch-source-equal-ordinal-test"

	payloadA := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Sent"}}`)
	payloadB := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Responded"}}`)

	idA, _, err := svc.Enqueue(ctx, "batch-source-equal-ordinal-test", "webhook.event", "ikey-batch-source-equal-a", payloadA, packInstall, 1, "", "", 0, "dlv-batch-source-equal-a")
	if err != nil {
		t.Fatalf("Enqueue shared-source first: %v", err)
	}
	idB, _, err := svc.Enqueue(ctx, "batch-source-equal-ordinal-test", "webhook.event", "ikey-batch-source-equal-b", payloadB, packInstall, 1, "", "", 0, "dlv-batch-source-equal-b")
	if err != nil {
		t.Fatalf("Enqueue shared-source second: %v", err)
	}

	if _, err := db.Exec(`UPDATE _System_ConnectorEvent
SET EnqueueOrdinal = 7
WHERE Id IN (?, ?)`, idA, idB); err != nil {
		t.Fatalf("force equal EnqueueOrdinal: %v", err)
	}

	peeked, err := svc.PeekBatch(ctx, packInstall, 10)
	if err != nil {
		t.Fatalf("PeekBatch initial: %v", err)
	}
	peekedIDs := make(map[string]bool)
	for _, ev := range peeked {
		peekedIDs[ev.ID] = true
	}
	if !peekedIDs[idA] {
		t.Fatalf("PeekBatch must include first equal-ordinal shared-source row %s; got %v", idA, peekedIDs)
	}
	if !peekedIDs[idB] {
		t.Fatalf("PeekBatch must include later equal-ordinal shared-source row %s; ClaimByID enforces ordering", idB)
	}

	tokenA, _, okA, err := svc.ClaimByID(ctx, idA, "worker-batch-source-a", packInstall, sql.NullString{String: "dlv-batch-source-equal-a", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID first equal-ordinal shared-source row: %v", err)
	}
	if !okA || tokenA == "" {
		t.Fatalf("ClaimByID first equal-ordinal shared-source row must win; got ok=%v token=%q", okA, tokenA)
	}

	if _, _, ok, err := svc.ClaimByID(ctx, idB, "worker-batch-source-b", packInstall, sql.NullString{String: "dlv-batch-source-equal-b", Valid: true}, 0); err != nil {
		t.Fatalf("ClaimByID blocked equal-ordinal shared-source sibling: %v", err)
	} else if ok {
		t.Fatalf("later equal-ordinal shared-source row %s must stay blocked while first is processing", idB)
	}

	if err := svc.MarkProcessed(ctx, idA, tokenA); err != nil {
		t.Fatalf("MarkProcessed first equal-ordinal shared-source row: %v", err)
	}

	tokenB, _, okB, err := svc.ClaimByID(ctx, idB, "worker-batch-source-b", packInstall, sql.NullString{String: "dlv-batch-source-equal-b", Valid: true}, 0)
	if err != nil {
		t.Fatalf("ClaimByID released equal-ordinal shared-source sibling: %v", err)
	}
	if !okB || tokenB == "" {
		t.Fatalf("later equal-ordinal shared-source row must become claimable after first is processed; got ok=%v token=%q", okB, tokenB)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// helpers used only by this file
// ─────────────────────────────────────────────────────────────────────────────

// allEventColumns returns the full-column list that PeekBatch SELECTs.
func allEventColumns() []string {
	return []string{
		"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
		"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
		"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
		"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey",
		"DeliveryId", "DeliverySequence", "CreatedDate", "LastModifiedDate",
	}
}

// eventRowValues returns one row of values in the same allEventColumns order.
// Returns []driver.Value so it can be spread into sqlmock AddRow(...driver.Value).
func eventRowValues(id, ns, packID, kind, ikey, status string, retryCount int, claimToken string) []driver.Value {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	return []driver.Value{
		id, ns, packID, kind, ikey, status,
		"default", []byte(`{}`), retryCount, nil, nil,
		nil, nil, claimToken, nil, nil, nil,
		int64(1), nil, nil, nil,
		nil, 0, now, now,
	}
}