package connector_workqueue

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

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

const replayOwnerQueryPattern = `(?s)SELECT w.WorkId, COALESCE\(w.ProjectionKey, ''\), w.State, w.ReplayOfWorkId, w.ReplayedAt, w.Attempt.*FROM _System_ConnectorWorkItem w.*w.EventId = \?.*w.WorkType IN \(\?, \?\).*CASE WHEN w.WorkType = \?.*LIMIT 2.*FOR UPDATE`
const replayOwnerCompatibilityQueryPattern = `(?s)SELECT w.WorkId, w.State, w.Attempt.*FROM _System_ConnectorWorkItem w.*w.EventId = \?.*w.WorkType IN \(\?, \?\).*CASE WHEN w.WorkType = \?.*LIMIT 2.*FOR UPDATE`

func TestRepositoryReserveDeadWorkItemReplayReservesClaimedDeadGeneration(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)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-dead", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-dead", "", WorkStateDead, nil, nil, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET ReplayedAt = \?.*WHERE WorkId = \?.*EventId = \?.*State = \?.*ReplayedAt IS NULL`).
		WithArgs(now, now, "work-dead", "event-dead", WorkStateDead).
		WillReturnResult(sqlmock.NewResult(0, 1))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-dead", WorkTypeConnectorEvent, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if !got.WorkItemOwned || !got.Reserved || got.WorkID != "work-dead" || got.ProjectionKey != "" {
		t.Fatalf("reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayUsesLegacyPathForZeroAttemptMirror(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, 5, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-legacy", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeWorkflowActionApply).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-legacy", "", WorkStateDead, nil, nil, 0))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-legacy", WorkTypeWorkflowActionApply, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if got.WorkItemOwned || got.Reserved || got.WorkID != "work-legacy" {
		t.Fatalf("legacy reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayFindsCompatibilityWorkTypeOwner(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, 7, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-action", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeWorkflowActionApply).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-compat-connector-event", "", WorkStateDead, nil, nil, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET ReplayedAt = \?.*WHERE WorkId = \?.*EventId = \?.*State = \?.*ReplayedAt IS NULL`).
		WithArgs(now, now, "work-compat-connector-event", "event-action", WorkStateDead).
		WillReturnResult(sqlmock.NewResult(0, 1))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-action", WorkTypeWorkflowActionApply, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if !got.WorkItemOwned || !got.Reserved || got.WorkID != "work-compat-connector-event" {
		t.Fatalf("compatibility reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayRejectsAmbiguousCompatibilityOwners(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, 8, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-ambiguous", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-connector", "", WorkStateDead, nil, nil, 1).
			AddRow("work-action", "", WorkStateDead, nil, nil, 1))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-ambiguous", WorkTypeConnectorEvent, now)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReserveDeadWorkItemReplay error = %v, want ErrReplayBlocked", err)
	}
	if got != (ReplayReservation{}) {
		t.Fatalf("ambiguous reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayBlocksZeroAttemptProjectionSuccessor(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, 9, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-projection-legacy", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-projection-legacy", "projection-legacy", WorkStateDead, nil, nil, 0))
	mock.ExpectQuery(`(?s)SELECT WorkId.*FROM _System_ConnectorWorkItem.*ProjectionActiveKey = \?.*WorkId <> \?.*State NOT IN \(\?, \?, \?\).*FOR UPDATE`).
		WithArgs("projection-legacy", "work-projection-legacy", WorkStateProcessed, WorkStateSkipped, WorkStateDead).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-projection-successor"))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-projection-legacy", WorkTypeConnectorEvent, now)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReserveDeadWorkItemReplay error = %v, want ErrReplayBlocked", err)
	}
	if got != (ReplayReservation{}) {
		t.Fatalf("blocked reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayRejectsConsumedOrNonterminalOwner(t *testing.T) {
	tests := []struct {
		name       string
		state      string
		replayedAt any
	}{
		{name: "nonterminal", state: WorkStateRetry, replayedAt: nil},
		{name: "already replayed", state: WorkStateDead, replayedAt: time.Date(2026, 7, 17, 10, 0, 0, 0, time.UTC)},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(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(replayOwnerQueryPattern).
				WithArgs("event-owned", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
				WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
					AddRow("work-owned", "", tt.state, nil, tt.replayedAt, 1))

			got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-owned", WorkTypeConnectorEvent, now)
			if err != nil {
				t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
			}
			if !got.WorkItemOwned || got.Reserved || got.WorkID != "work-owned" {
				t.Fatalf("reservation = %+v", got)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sqlmock expectations: %v", err)
			}
		})
	}
}

func TestRepositoryReserveDeadWorkItemReplayBlocksActiveProjectionSuccessor(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(replayOwnerQueryPattern).
		WithArgs("event-projection-dead", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-projection-dead", "projection-1", WorkStateDead, nil, nil, 1))
	mock.ExpectQuery(`(?s)SELECT WorkId.*FROM _System_ConnectorWorkItem.*ProjectionActiveKey = \?.*WorkId <> \?.*State NOT IN \(\?, \?, \?\).*FOR UPDATE`).
		WithArgs("projection-1", "work-projection-dead", WorkStateProcessed, WorkStateSkipped, WorkStateDead).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-projection-active"))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-projection-dead", WorkTypeConnectorEvent, now)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReserveDeadWorkItemReplay error = %v, want ErrReplayBlocked", err)
	}
	if got != (ReplayReservation{}) {
		t.Fatalf("blocked reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryFenceZeroAttemptReplayMirrorTerminalizesClaimableState(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, 18, 0, 0, time.UTC)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*TerminalAt = COALESCE\(TerminalAt, \?\).*ReplayedAt = \?.*ProjectionActiveKey = NULL.*WHERE WorkId = \?.*EventId = \?.*Attempt = 0.*State IN \(\?, \?, \?, \?, \?\).*ReplayedAt IS NULL`).
		WithArgs(
			WorkStateDead, now, now, now,
			"work-zero", "event-zero",
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

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

func TestRepositoryFenceZeroAttemptReplayMirrorRejectsConflictingState(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, 19, 0, 0, time.UTC)
	mock.ExpectExec(`UPDATE _System_ConnectorWorkItem`).
		WillReturnResult(sqlmock.NewResult(0, 0))

	fenced, err := NewRepository(db, nil, nil).FenceZeroAttemptReplayMirror(
		context.Background(), "event-zero", "work-zero", now,
	)
	if err != nil || fenced {
		t.Fatalf("FenceZeroAttemptReplayMirror = (%v, %v), want false, nil", fenced, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryFenceZeroAttemptReplayMirrorFallsBackBeforeTerminalAtSchema(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, 19, 30, 0, time.UTC)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = COALESCE\(TerminalAt, \?\).*ProjectionActiveKey = NULL.*WHERE WorkId = \?`).
		WithArgs(
			WorkStateDead, now, now, now,
			"work-zero", "event-zero",
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'TerminalAt' in 'field list'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*ReplayedAt = \?.*ProjectionActiveKey = NULL.*LastModifiedDate = \?.*WHERE WorkId = \?`).
		WithArgs(
			WorkStateDead, now, now,
			"work-zero", "event-zero",
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

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

func TestRepositoryFenceZeroAttemptReplayMirrorFallsBackBeforeProjectionActiveKeySchema(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, 19, 45, 0, time.UTC)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = COALESCE\(TerminalAt, \?\).*ProjectionActiveKey = NULL.*WHERE WorkId = \?`).
		WithArgs(
			WorkStateDead, now, now, now,
			"work-zero", "event-zero",
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'ProjectionActiveKey' in 'field list'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = COALESCE\(TerminalAt, \?\).*ReplayedAt = \?.*LastModifiedDate = \?.*WHERE WorkId = \?`).
		WithArgs(
			WorkStateDead, now, now, now,
			"work-zero", "event-zero",
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

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

func TestRepositoryReserveDeadWorkItemReplayFallsBackForOlderSchema(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, 20, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-old-schema", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'w.ProjectionKey' in 'field list'"))
	mock.ExpectQuery(replayOwnerCompatibilityQueryPattern).
		WithArgs("event-old-schema", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "State", "Attempt"}).
			AddRow("work-old-schema", WorkStateBlocked, 0))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-old-schema", WorkTypeConnectorEvent, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if got.WorkItemOwned || got.Reserved || got.WorkID != "work-old-schema" || got.ProjectionKey != "" {
		t.Fatalf("older-schema reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayPreservesEventOnlyFallbackWithoutWorkItemTable(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, 21, 0, 0, time.UTC)
	tableMissing := errors.New("Error 1146 (42S02): Table 'nexuscrm._System_ConnectorWorkItem' doesn't exist")
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-only", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnError(tableMissing)
	mock.ExpectQuery(replayOwnerCompatibilityQueryPattern).
		WithArgs("event-only", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnError(tableMissing)

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-only", WorkTypeConnectorEvent, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if got != (ReplayReservation{}) {
		t.Fatalf("event-only reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayFallsBackWithoutReplayLineageColumn(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, 22, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-partial-schema", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'w.ReplayOfWorkId' in 'field list'"))
	mock.ExpectQuery(replayOwnerCompatibilityQueryPattern).
		WithArgs("event-partial-schema", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "State", "Attempt"}).
			AddRow("work-partial-schema", WorkStateDead, 1))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(context.Background(), "event-partial-schema", WorkTypeConnectorEvent, now)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if !got.WorkItemOwned || got.Reserved || got.WorkID != "work-partial-schema" || got.ProjectionKey != "" {
		t.Fatalf("partial-schema reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryReserveDeadWorkItemReplayFallsBackWithoutProjectionActiveKeyColumn(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, 24, 0, 0, time.UTC)
	mock.ExpectQuery(replayOwnerQueryPattern).
		WithArgs("event-partial-projection-schema", WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow("work-zero", "projection-1", WorkStateBlocked, nil, nil, 0))
	mock.ExpectQuery(`(?s)SELECT WorkId.*FROM _System_ConnectorWorkItem.*ProjectionActiveKey = \?.*FOR UPDATE`).
		WithArgs("projection-1", "work-zero", WorkStateProcessed, WorkStateSkipped, WorkStateDead).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'ProjectionActiveKey' in 'where clause'"))

	got, err := NewRepository(db, nil, nil).ReserveDeadWorkItemReplay(
		context.Background(),
		"event-partial-projection-schema",
		WorkTypeConnectorEvent,
		now,
	)
	if err != nil {
		t.Fatalf("ReserveDeadWorkItemReplay: %v", err)
	}
	if got.WorkItemOwned || got.Reserved || got.WorkID != "work-zero" || got.ProjectionKey != "projection-1" {
		t.Fatalf("partial projection schema reservation = %+v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryEnsureWorkItemStoresReplayLineage(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, 25, 0, 0, time.UTC)
	runAfter := now.Add(time.Second)
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkItem.*ReplayOfWorkId.*VALUES`).
		WithArgs(
			"work-replay", "event-replay", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
			WorkStatePending, runAfter, "", "", int64(1), int64(0), "work-original", now, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	created, err := NewRepository(db, nil, nil).EnsureReplayWorkItem(context.Background(), ReplayWorkItemInput{
		WorkItem: CreateWorkItemInput{
			WorkID:          "work-replay",
			EventID:         "event-replay",
			InstalledPackID: "pack-1",
			WorkType:        WorkTypeConnectorEvent,
			QueueClass:      DefaultQueueClass,
			State:           WorkStatePending,
			RunAfter:        runAfter,
			LaneOrdinal:     1,
			Now:             now,
		},
		ReplayOfWorkID: " work-original ",
	})
	if err != nil {
		t.Fatalf("EnsureWorkItem: %v", err)
	}
	if !created {
		t.Fatal("EnsureWorkItem created=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryRejectsSelfReferentialReplayLineage(t *testing.T) {
	now := time.Date(2026, 7, 17, 11, 35, 0, 0, time.UTC)
	_, err := NewRepository(nil, nil, nil).EnsureReplayWorkItem(context.Background(), ReplayWorkItemInput{
		WorkItem: CreateWorkItemInput{
			WorkID:          "work-self",
			EventID:         "event-self",
			InstalledPackID: "pack-1",
			WorkType:        WorkTypeConnectorEvent,
			QueueClass:      DefaultQueueClass,
			State:           WorkStatePending,
			RunAfter:        now,
			Now:             now,
		},
		ReplayOfWorkID: "work-self",
	})
	if !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("EnsureWorkItem error = %v, want ErrInvalidWorkItem", err)
	}
}