package connector_workqueue

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
)

func ordinaryWorkItemIdentityQueryPattern(schema workItemIdentityReadSchema) string {
	return regexp.QuoteMeta(ordinaryWorkItemIdentityQuery(schema))
}

func ordinaryWorkItemIdentityRows(schema workItemIdentityReadSchema) *sqlmock.Rows {
	columns := []string{
		"WorkId",
		"EventId",
		"InstalledPackId",
		"WorkType",
		"QueueClass",
		"SourceLaneKey",
		"TargetMutationLaneKey",
		"LaneOrdinal",
		"Generation",
	}
	if schema.replayOfWorkID {
		columns = append(columns, "ReplayOfWorkId")
	}
	if schema.projectionKey {
		columns = append(columns, "ProjectionKey")
	}
	if schema.projectionActiveKey {
		columns = append(columns, "ProjectionActiveKey")
	}
	return sqlmock.NewRows(columns)
}

func ordinaryWorkItemIdentityRow(
	schema workItemIdentityReadSchema,
	workID, eventID, installedPackID, workType, queueClass string,
	sourceLaneKey, targetMutationLaneKey string,
	laneOrdinal, generation int64,
	replayOfWorkID, projectionKey, projectionActiveKey string,
) []driver.Value {
	values := []driver.Value{
		workID,
		eventID,
		installedPackID,
		workType,
		queueClass,
		sourceLaneKey,
		targetMutationLaneKey,
		laneOrdinal,
		generation,
	}
	if schema.replayOfWorkID {
		values = append(values, replayOfWorkID)
	}
	if schema.projectionKey {
		values = append(values, projectionKey)
	}
	if schema.projectionActiveKey {
		values = append(values, projectionActiveKey)
	}
	return values
}

func TestRepositoryEnsureWorkItemConvergesExistingImmutableIdentity(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, 31, 9, 0, 0, 0, time.UTC)
	schema := allWorkItemIdentityColumns()
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate EventId and WorkType"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(schema)).
		WithArgs("work-proposed", "event-1", WorkTypeConnectorEvent).
		WillReturnRows(ordinaryWorkItemIdentityRows(schema).AddRow(ordinaryWorkItemIdentityRow(
			schema,
			"work-existing", "event-1", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
			"source:1", "", 7, 3, "", "", "",
		)...))

	created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID: "work-proposed", EventID: "event-1", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
		QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now,
		SourceLaneKey: "source:1", LaneOrdinal: 7, Generation: 3, Now: now,
	})
	if err != nil {
		t.Fatalf("EnsureWorkItem: %v", err)
	}
	if created {
		t.Fatal("EnsureWorkItem created=true, want convergence to existing row")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryEnsureWorkItemFreshInsertDoesNotReadOptionalIdentitySchema(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, 31, 9, 5, 0, 0, time.UTC)
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID: "work-new", EventID: "event-new", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
		QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now, Now: now,
	})
	if err != nil || !created {
		t.Fatalf("EnsureWorkItem = (%v, %v), want fresh insert without identity read", created, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryEnsureWorkItemRollingSchemaFallsBackPerMissingOptionalColumn(t *testing.T) {
	for _, missing := range []string{"ReplayOfWorkId", "ProjectionKey", "ProjectionActiveKey"} {
		t.Run(missing, 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, 31, 9, 10, 0, 0, time.UTC)
			fullSchema := allWorkItemIdentityColumns()
			fallbackSchema := fullSchema
			switch missing {
			case "ReplayOfWorkId":
				fallbackSchema.replayOfWorkID = false
			case "ProjectionKey":
				fallbackSchema.projectionKey = false
			case "ProjectionActiveKey":
				fallbackSchema.projectionActiveKey = false
			}

			mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
				WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate EventId and WorkType"})
			mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(fullSchema)).
				WithArgs("work-new", "event-new", WorkTypeConnectorEvent).
				WillReturnError(&mysql.MySQLError{Number: 1054, Message: "Unknown column '" + missing + "' in 'field list'"})
			mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(fallbackSchema)).
				WithArgs("work-new", "event-new", WorkTypeConnectorEvent).
				WillReturnRows(ordinaryWorkItemIdentityRows(fallbackSchema).AddRow(ordinaryWorkItemIdentityRow(
					fallbackSchema,
					"work-existing", "event-new", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
					"", "", 1, 0, "", "", "",
				)...))

			created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
				WorkID: "work-new", EventID: "event-new", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
				QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now, LaneOrdinal: 1, Now: now,
			})
			if err != nil {
				t.Fatalf("EnsureWorkItem with missing %s: %v", missing, err)
			}
			if created {
				t.Fatalf("EnsureWorkItem with missing %s created=true, want existing-row convergence", missing)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sqlmock expectations: %v", err)
			}
		})
	}
}

func TestRepositoryEnsureWorkItemRollingSchemaStillChecksPresentOptionalIdentity(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, 31, 9, 20, 0, 0, time.UTC)
	fullSchema := allWorkItemIdentityColumns()
	fallbackSchema := fullSchema
	fallbackSchema.replayOfWorkID = false
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate WorkId"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(fullSchema)).
		WithArgs("work-1", "event-1", WorkTypeConnectorEvent).
		WillReturnError(&mysql.MySQLError{Number: 1054, Message: "Unknown column 'ReplayOfWorkId' in 'field list'"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(fallbackSchema)).
		WithArgs("work-1", "event-1", WorkTypeConnectorEvent).
		WillReturnRows(ordinaryWorkItemIdentityRows(fallbackSchema).AddRow(ordinaryWorkItemIdentityRow(
			fallbackSchema,
			"work-1", "event-1", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
			"", "", 1, 0, "", "csp:projection", "csp:projection",
		)...))

	created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID: "work-1", EventID: "event-1", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
		QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now, LaneOrdinal: 1, Now: now,
	})
	if !errors.Is(err, ErrWorkItemIdentityConflict) {
		t.Fatalf("EnsureWorkItem error = %v, want present projection identity conflict", err)
	}
	if created {
		t.Fatal("EnsureWorkItem created=true for projection-owned identity")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryEnsureWorkItemDoesNotHideRequiredSchemaFailures(t *testing.T) {
	for _, tc := range []struct {
		name string
		err  error
	}{
		{name: "required column", err: &mysql.MySQLError{Number: 1054, Message: "Unknown column 'LaneOrdinal' in 'field list'"}},
		{name: "missing table", err: &mysql.MySQLError{Number: 1146, Message: "Table '_System_ConnectorWorkItem' doesn't exist"}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock.New: %v", err)
			}
			defer db.Close()

			schema := allWorkItemIdentityColumns()
			mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
				WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate WorkId"})
			mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(schema)).
				WithArgs("work-1", "event-1", WorkTypeConnectorEvent).
				WillReturnError(tc.err)

			_, err = NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
				WorkID: "work-1", EventID: "event-1", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
			})
			if !errors.Is(err, tc.err) {
				t.Fatalf("EnsureWorkItem error = %v, want original schema failure", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sqlmock expectations: %v", err)
			}
		})
	}
}

func TestRepositoryEnsureWorkItemRejectsCrossedDurableIdentities(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, 31, 9, 30, 0, 0, time.UTC)
	schema := allWorkItemIdentityColumns()
	rows := ordinaryWorkItemIdentityRows(schema)
	rows.AddRow(ordinaryWorkItemIdentityRow(
		schema,
		"work-a", "event-other", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
		"", "", 1, 0, "", "", "",
	)...)
	rows.AddRow(ordinaryWorkItemIdentityRow(
		schema,
		"work-b", "event-1", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass,
		"", "", 1, 0, "", "", "",
	)...)
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate identities"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(schema)).
		WithArgs("work-a", "event-1", WorkTypeConnectorEvent).
		WillReturnRows(rows)

	created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID: "work-a", EventID: "event-1", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
		QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now, LaneOrdinal: 1, Now: now,
	})
	if !errors.Is(err, ErrWorkItemIdentityConflict) {
		t.Fatalf("EnsureWorkItem error = %v, want crossed identity conflict", err)
	}
	if created {
		t.Fatal("EnsureWorkItem created=true for crossed identities")
	}
	var conflict *WorkItemIdentityConflictError
	if !errors.As(err, &conflict) || conflict.ExistingWorkID != "work-b" {
		t.Fatalf("EnsureWorkItem conflict = %+v, want existing work-b", conflict)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryEnsureWorkItemRejectsPersistedImmutableMismatch(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, 31, 9, 40, 0, 0, time.UTC)
	schema := allWorkItemIdentityColumns()
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "duplicate WorkId"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(schema)).
		WithArgs("work-1", "event-1", WorkTypeConnectorEvent).
		WillReturnRows(ordinaryWorkItemIdentityRows(schema).AddRow(ordinaryWorkItemIdentityRow(
			schema,
			"work-1", "event-1", "pack-other", WorkTypeConnectorEvent, DefaultQueueClass,
			"", "", 1, 0, "", "", "",
		)...))

	created, err := NewRepository(db, nil, nil).EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID: "work-1", EventID: "event-1", InstalledPackID: "pack-1", WorkType: WorkTypeConnectorEvent,
		QueueClass: DefaultQueueClass, State: WorkStatePending, RunAfter: now, LaneOrdinal: 1, Now: now,
	})
	if !errors.Is(err, ErrWorkItemIdentityConflict) {
		t.Fatalf("EnsureWorkItem error = %v, want immutable field conflict", err)
	}
	if created {
		t.Fatal("EnsureWorkItem created=true for immutable field mismatch")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestIsDuplicateKeyRequiresTypedMySQL1062(t *testing.T) {
	if isDuplicateKey(errors.New("Error 1062: Duplicate entry")) {
		t.Fatal("string-shaped error classified as MySQL duplicate")
	}
	if !isDuplicateKey(&mysql.MySQLError{Number: 1062, Message: "duplicate"}) {
		t.Fatal("typed MySQL 1062 was not classified as duplicate")
	}
	if isDuplicateKey(&mysql.MySQLError{Number: 1213, Message: "deadlock"}) {
		t.Fatal("non-1062 MySQL error classified as duplicate")
	}
}