package connector_workqueue

import (
	"context"
	"errors"
	"regexp"
	"testing"
	"time"

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

func TestRepositoryCompleteExecutorWorkItemForCoalescingUsesLeaseAndOwnershipEpochCAS(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 17, 10, 0, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT OwnershipEpoch, WorkType.*WHERE WorkId = \?.*AND EventId = \?.*AND Generation = \?.*AND State = \?.*AND LeaseToken = \?.*FOR UPDATE`).
		WithArgs("work-current", "event-current", int64(7), WorkStateProcessing, "lease-current", SourceProjectionWorkIDPrefix+"%").
		WillReturnRows(sqlmock.NewRows([]string{"OwnershipEpoch", "WorkType"}).AddRow(int64(11), WorkTypeConnectorEvent))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*WHERE WorkId = \?.*AND EventId = \?.*AND WorkType = \?.*AND Generation = \?.*AND State = \?.*AND LeaseToken = \?.*AND OwnershipEpoch = \?.*AND WorkId NOT LIKE \?`).
		WithArgs(WorkStateProcessed, now, now, "work-current", "event-current", WorkTypeConnectorEvent, int64(7), WorkStateProcessing, "lease-current", int64(11), SourceProjectionWorkIDPrefix+"%").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-current").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "", ""))

	completed, err := NewRepository(db, nil, nil).CompleteExecutorWorkItemForCoalescing(
		context.Background(), "work-current", "event-current", "lease-current", 7, now,
	)
	if err != nil || !completed {
		t.Fatalf("CompleteExecutorWorkItemForCoalescing = (%v, %v), want true, nil", completed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryCompleteExecutorWorkItemForCoalescingFallsBackBeforeTerminalAtSchema(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 17, 10, 15, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT OwnershipEpoch, WorkType.*WHERE WorkId = \?.*AND EventId = \?.*AND Generation = \?.*AND State = \?.*AND LeaseToken = \?.*FOR UPDATE`).
		WithArgs("work-legacy", "event-legacy", int64(7), WorkStateProcessing, "lease-legacy", SourceProjectionWorkIDPrefix+"%").
		WillReturnRows(sqlmock.NewRows([]string{"OwnershipEpoch", "WorkType"}).AddRow(int64(12), WorkTypeConnectorEvent))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = \?.*WHERE WorkId = \?.*AND OwnershipEpoch = \?`).
		WithArgs(WorkStateProcessed, now, now, "work-legacy", "event-legacy", WorkTypeConnectorEvent, int64(7), WorkStateProcessing, "lease-legacy", int64(12), SourceProjectionWorkIDPrefix+"%").
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'TerminalAt' in 'field list'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*LastModifiedDate = \?.*WHERE WorkId = \?.*AND OwnershipEpoch = \?`).
		WithArgs(WorkStateProcessed, now, "work-legacy", "event-legacy", WorkTypeConnectorEvent, int64(7), WorkStateProcessing, "lease-legacy", int64(12), SourceProjectionWorkIDPrefix+"%").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-legacy").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "", ""))

	completed, err := NewRepository(db, nil, nil).CompleteExecutorWorkItemForCoalescing(
		context.Background(), "work-legacy", "event-legacy", "lease-legacy", 7, now,
	)
	if err != nil || !completed {
		t.Fatalf("legacy-schema coalescing completion = (%v, %v), want true, nil", completed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryCompleteExecutorWorkItemForCoalescingRejectsSourceProjection(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	completed, err := NewRepository(db, nil, nil).CompleteExecutorWorkItemForCoalescing(
		context.Background(), SourceProjectionWorkIDPrefix+"projection", "event-current",
		"lease-current", 7, time.Now().UTC(),
	)
	if completed || !errors.Is(err, ErrInvalidTerminalState) {
		t.Fatalf("source projection completion = (%v, %v), want false, ErrInvalidTerminalState", completed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected database access: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventForCoalescingHonorsClaimWorkTypes(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Now().UTC()
	ctx := WithClaimWorkTypes(context.Background(), []string{WorkTypeConnectorEvent})
	work, claimed, err := NewRepository(db, nil, nil).ClaimWorkItemByEventForCoalescing(ctx, ClaimWorkItemByEventInput{
		EventID:         "event-next",
		WorkType:        WorkTypeWorkflowActionApply,
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-next",
		LeaseUntil:      now.Add(time.Minute),
		Now:             now,
	}, false)
	if work != nil || claimed || err != nil {
		t.Fatalf("scoped coalescing claim = (%+v, %v, %v), want nil, false, nil", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected database access: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventForCoalescingUsesStoredCompatibilityWorkType(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 17, 11, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(time.Minute)
	mock.ExpectQuery(`(?s)SELECT w.WorkId, w.WorkType, w.OwnershipEpoch.*w.WorkType IN \(\?, \?\).*w.WorkId NOT LIKE \?.*ProjectionKey.*ProjectionActiveKey.*e.ExecutionOwner.*LIMIT 2.*FOR UPDATE`).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "WorkType", "OwnershipEpoch"}).
			AddRow("work-next", WorkTypeConnectorEvent, int64(8)))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem AS w.*LEFT JOIN _System_ConnectorEvent AS e.*SET w.State = \?.*w.OwnershipEpoch = w.OwnershipEpoch \+ 1.*e.Status = CASE WHEN e.ExecutionOwner = \?.*e.ExecutionOwner = \?.*WHERE w.WorkId = \?.*AND w.OwnershipEpoch = \?.*ProjectionKey.*ProjectionActiveKey.*e.ExecutionOwner`).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-next").
		WillReturnRows(workItemRows().AddRow(
			"work-next", "event-next", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
			WorkStateProcessing, now, 2, "owner-1", "lease-next", leaseUntil,
			"", "", int64(2), int64(7), now, now,
		))

	ctx := WithDistributedClaim(WithClaimWorkTypes(context.Background(), []string{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
	}))
	work, claimed, err := NewRepository(db, nil, nil).ClaimWorkItemByEventForCoalescing(ctx, ClaimWorkItemByEventInput{
		EventID:         "event-next",
		WorkType:        WorkTypeWorkflowActionApply,
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-next",
		LeaseUntil:      leaseUntil,
		Now:             now,
	}, true)
	if err != nil || !claimed || work == nil || work.WorkID != "work-next" || work.WorkType != WorkTypeConnectorEvent {
		t.Fatalf("compatibility coalescing claim = (%+v, %v, %v), want stored connector_event work-next, true, nil", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventForCoalescingLosesOwnershipEpochCAS(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 17, 11, 10, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT w.WorkId, w.WorkType, w.OwnershipEpoch.*w.WorkType IN \(\?, \?\).*FOR UPDATE`).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "WorkType", "OwnershipEpoch"}).
			AddRow("work-next", WorkTypeConnectorEvent, int64(8)))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem AS w.*WHERE w.WorkId = \?.*AND w.OwnershipEpoch = \?.*e.ExecutionOwner`).
		WillReturnResult(sqlmock.NewResult(0, 0))

	ctx := WithDistributedClaim(WithClaimWorkTypes(context.Background(), []string{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
	}))
	work, claimed, err := NewRepository(db, nil, nil).ClaimWorkItemByEventForCoalescing(ctx, ClaimWorkItemByEventInput{
		EventID:         "event-next",
		WorkType:        WorkTypeWorkflowActionApply,
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-next",
		LeaseUntil:      now.Add(time.Minute),
		Now:             now,
	}, true)
	if work != nil || claimed || err != nil {
		t.Fatalf("lost epoch coalescing claim = (%+v, %v, %v), want nil, false, nil", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventForCoalescingRejectsAmbiguousExecutorRows(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 17, 11, 15, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT w.WorkId, w.WorkType, w.OwnershipEpoch.*w.WorkType IN \(\?, \?\).*LIMIT 2.*FOR UPDATE`).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "WorkType", "OwnershipEpoch"}).
			AddRow("work-connector", WorkTypeConnectorEvent, int64(8)).
			AddRow("work-action", WorkTypeWorkflowActionApply, int64(9)))

	ctx := WithDistributedClaim(WithClaimWorkTypes(context.Background(), []string{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
	}))
	work, claimed, err := NewRepository(db, nil, nil).ClaimWorkItemByEventForCoalescing(ctx, ClaimWorkItemByEventInput{
		EventID:         "event-ambiguous",
		WorkType:        WorkTypeWorkflowActionApply,
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-next",
		LeaseUntil:      now.Add(time.Minute),
		Now:             now,
	}, true)
	if work != nil || claimed || !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("ambiguous compatibility coalescing claim = (%+v, %v, %v), want nil, false, ErrInvalidWorkItem", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}