package packs

import (
	"bytes"
	"context"
	"database/sql"
	"database/sql/driver"
	"fmt"
	"log"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
)

func TestBackfillPolymorphicDiscriminatorsUpdatesMissingAndWrongValues(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{}).
			AddRow("04i-2", "00G-group", sql.NullString{String: "User", Valid: true}).
			AddRow("04i-3", "005-user", sql.NullString{String: "Group", Valid: true}))

	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 2))).
		WithArgs("005-user", "00G-group").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 2))).
		WithArgs("005-user", "00G-group").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("00G-group"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 2))).
		WithArgs("User", "04i-1", "005-user", nil, "04i-3", "005-user", "Group").
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("Group", "04i-2", "00G-group", "User").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFindsRepairRowsWithBoundedCandidateProbes(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{}).
			AddRow("04i-3", "005-user", sql.NullString{String: "Group", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-2", "00G-group", sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 2))).
		WithArgs("005-user", "00G-group").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 2))).
		WithArgs("005-user", "00G-group").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("00G-group"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 2))).
		WithArgs("User", "04i-1", "005-user", nil, "04i-3", "005-user", "Group").
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("Group", "04i-2", "00G-group", "User").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbesSkipCorrectRowsWithoutSourcePaging(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectEmptyPolyBackfillCandidateProbes(mock)

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbesClearStaleType(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", nil, sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs(nil, "04i-1", nil, "User").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbeFindsLateRepairWithoutPagingCorrectRows(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1001", "005-user", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("User", "04i-1001", "005-user", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbesLeaveUnresolvedFKUnchanged(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectEmptyPolyBackfillCandidateProbes(mock)

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbesFailClosedOnAmbiguousFK(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	for _, target := range []string{"User", "Group"} {
		mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL(target, false))).
			WithArgs(target).
			WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
				AddRow("04i-1", "shared-id", sql.NullString{}))
	}
	for _, target := range []string{"User", "Group"} {
		mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL(target, 1))).
			WithArgs("shared-id").
			WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("shared-id"))
	}

	err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()})
	if err == nil || !strings.Contains(err.Error(), "matched targets=[User Group]") {
		t.Fatalf("error = %v, want ambiguity diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbeFailsClosedOnEmptySourceID(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillInvalidSourceIDSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("", "005-user", sql.NullString{String: "User", Valid: true}))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()})
	if err == nil || !strings.Contains(err.Error(), "encountered empty row Id") {
		t.Fatalf("error = %v, want empty source Id diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsCandidateProbeFailsClosedOnDuplicateSourceID(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	candidateRows := sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
		AddRow("04i-1", nil, sql.NullString{String: "User", Valid: true}).
		AddRow("04i-1", nil, sql.NullString{String: "User", Valid: true})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(candidateRows)

	err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()})
	if err == nil || !strings.Contains(err.Error(), `encountered duplicate row Id="04i-1"`) {
		t.Fatalf("error = %v, want duplicate source Id diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFallsBackWhenCandidateTargetProbeIsUnavailable(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.Group' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("005-user").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.Group' doesn't exist"})

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsRestartsCanonicalSourcePagingAfterFullCandidatePage(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	firstPage := sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"})
	for i := 0; i < polymorphicBackfillBatchSize; i++ {
		rowID := fmt.Sprintf("04i-%03d", i)
		fkID := fmt.Sprintf("005-%03d", i)
		firstPage.AddRow(rowID, fkID, sql.NullString{})
	}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(firstPage)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	// The candidate discovery page is deliberately discarded. Canonical source
	// paging restarts at the beginning instead of advancing with a Go-sorted
	// cursor that may disagree with the database collation.
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillCombinedSourceSelectSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type", "UserId", "GroupId"}).
			AddRow("04i-500", "00G-group", sql.NullString{}, nil, "00G-group"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("Group", "04i-500", "00G-group", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{polyBackfillTestDef()}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsIsIdempotent(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsClearsStaleTypeForNullFK(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", nil, sql.NullString{String: "User", Valid: true}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs(nil, "04i-1", nil, "User").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsClearsStaleTypeForEmptyFK(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "", sql.NullString{String: "Group", Valid: true}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs(nil, "04i-1", "", "Group").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsLeavesEmptyFKWithEmptyType(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", nil, sql.NullString{String: "", Valid: true}))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsLeavesUnresolvedHistoricalFKUnchanged(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()
	var logs bytes.Buffer
	previousLogWriter := log.Writer()
	log.SetOutput(&logs)
	t.Cleanup(func() { log.SetOutput(previousLogWriter) })

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "missing", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("missing").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("missing").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if !strings.Contains(logs.String(), "unresolved=1") {
		t.Fatalf("logs = %q, want unresolved row count", logs.String())
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsRepairsResolvableRowsBesideUnresolvedHistoricalRows(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "missing", sql.NullString{}).
			AddRow("04i-2", "005-user", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 2))).
		WithArgs("missing", "005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 2))).
		WithArgs("missing", "005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("User", "04i-2", "005-user", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsAllowsUnavailableTargetWhenEveryFKResolvesElsewhere(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "006-oppty", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("006-oppty").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("006-oppty").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("006-oppty"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("EmailMessage", "RelatedToId", "RelatedToId_type", 1))).
		WithArgs("Opportunity", "02s-1", "006-oppty", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsLeavesUnresolvedFKWhenUnavailableTargetIsSourceExternal(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()
	var logs bytes.Buffer
	previousLogWriter := log.Writer()
	log.SetOutput(&logs)
	t.Cleanup(func() { log.SetOutput(previousLogWriter) })

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "a0G5g000000LphREAS", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G5g000000LphREAS").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G5g000000LphREAS").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("a0G", "Month_End_ARR__c"))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if !strings.Contains(logs.String(), "unresolved=1") || !strings.Contains(logs.String(), "unavailable_targets=1") {
		t.Fatalf("logs = %q, want unresolved row and unavailable target counts", logs.String())
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsLeavesHistoricalFKWhenUnavailableTargetCannotOwnIt(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "006-deleted", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("006-deleted").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("006-deleted").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("006", "Opportunity"))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedWhenUnavailableTargetOwnsUnresolvedFK(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "a0G-target", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G-target").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G-target").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("a0G", "AppointmentCategory"))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want unavailable target ownership error")
	}
	for _, want := range []string{"a0G-target", `owner="AppointmentCategory"`, "declared target schema is unavailable"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("error = %v, want diagnostic %q", err, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedWhenAllTargetSchemasAreUnavailable(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.User' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("005-user").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.Group' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("005", "User"))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want unavailable target schema error")
	}
	for _, want := range []string{"005-user", "targets=[User Group]", "schema is unavailable"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("error = %v, want diagnostic %q", err, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedWhenUnavailableTargetLeavesUnknownOwner(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "zzz-unknown", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("zzz-unknown").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("zzz-unknown").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want unknown ownership error")
	}
	for _, want := range []string{"zzz-unknown", `prefix="zzz"`, "active object ownership is unknown"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("error = %v, want diagnostic %q", err, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedOnUnknownIdColumn(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "a0G-external", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G-external").
		WillReturnError(&mysql.MySQLError{Number: 1054, Message: "Unknown column 'matched.Id' in 'on clause'"})

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want unknown-column schema error")
	}
	for _, want := range []string{"check target AppointmentCategory", "Unknown column"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("error = %v, want diagnostic %q", err, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedOnAmbiguousKeyPrefixOwnership(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "a0G-external", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G-external").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G-external").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("a0G", "Month_End_ARR__c").
			AddRow("a0G", "Other_Extension__c"))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want ambiguous key-prefix ownership error")
	}
	if !strings.Contains(err.Error(), `ambiguous active object key-prefix ownership prefix="a0G"`) {
		t.Fatalf("error = %v, want ambiguous key-prefix ownership diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedWhenKeyPrefixRegistryCannotBeLoaded(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillRelatedToTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillRelatedToSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RelatedToId", "RelatedToId_type"}).
			AddRow("02s-1", "a0G-external", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G-external").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G-external").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnError(&mysql.MySQLError{Number: 1142, Message: "SELECT command denied"})

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want key-prefix registry query error")
	}
	for _, want := range []string{"load object ownership", "SELECT command denied"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("error = %v, want diagnostic %q", err, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestResolvePolymorphicDiscriminatorTargetsCachesUnavailableTargetsAndObjectOwners(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	state := &polymorphicBackfillResolutionState{unavailableTargetErrors: make(map[string]error)}
	targets := []string{"AppointmentCategory", "Opportunity"}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("AppointmentCategory", 1))).
		WithArgs("a0G-first").
		WillReturnError(&mysql.MySQLError{Number: 1146, Message: "Table 'nexuscrm.AppointmentCategory' doesn't exist"})
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G-first").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillObjectKeyPrefixOwnersSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"KeyPrefix", "ApiName"}).
			AddRow("a0G", "Month_End_ARR__c"))

	firstResolved, err := resolvePolymorphicDiscriminatorTargets(
		context.Background(), db, targets,
		[]polymorphicBackfillRow{{id: "02s-1", fkID: sql.NullString{String: "a0G-first", Valid: true}}},
		state,
	)
	if err != nil {
		t.Fatalf("first resolvePolymorphicDiscriminatorTargets: %v", err)
	}
	if len(firstResolved["a0G-first"]) != 0 {
		t.Fatalf("first resolved targets = %v, want none", firstResolved)
	}

	// The second resolution must neither probe the known-missing target nor
	// reload the prefix registry, while still applying the unavailable-target
	// ownership guard to the new unresolved FK.
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Opportunity", 1))).
		WithArgs("a0G-second").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	secondResolved, err := resolvePolymorphicDiscriminatorTargets(
		context.Background(), db, targets,
		[]polymorphicBackfillRow{{id: "02s-2", fkID: sql.NullString{String: "a0G-second", Valid: true}}},
		state,
	)
	if err != nil {
		t.Fatalf("second resolvePolymorphicDiscriminatorTargets: %v", err)
	}
	if len(secondResolved["a0G-second"]) != 0 {
		t.Fatalf("second resolved targets = %v, want none", secondResolved)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsResolvesStoredGroupMemberGroupRows(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	groupMember := findSystemTableForPolyTest(t, systemTableDefinitions(), constants.TableGroupMember)
	mock.ExpectQuery(regexp.QuoteMeta(groupMemberPolyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "UserOrGroupId", "UserOrGroupId_type"}).
			AddRow("011-1", "00G-nested", sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("00G-nested").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("00G-nested").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("00G-nested"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("GroupMember", "UserOrGroupId", "UserOrGroupId_type", 1))).
		WithArgs("Group", "011-1", "00G-nested", "User").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{groupMember}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsUsesDatabaseCollationForTargetMatch(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "00g-group", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("00g-group").
		WillReturnRows(sqlmock.NewRows([]string{"RequestedId"}))
	// The joined Group row may store 00G-group, but SQL selects the original
	// requested value so the resolver maps the match back to the source FK.
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("00g-group").
		WillReturnRows(sqlmock.NewRows([]string{"RequestedId"}).AddRow("00g-group"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("Group", "04i-1", "00g-group", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedOnAmbiguousFK(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "shared-id", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("shared-id").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("shared-id"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("shared-id").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("shared-id"))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want ambiguous FK error")
	}
	if !strings.Contains(err.Error(), "ambiguous") || !strings.Contains(err.Error(), "matched targets=[User Group]") {
		t.Fatalf("error = %v, want ambiguity diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsFailsClosedOnConcurrentSourceChange(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("User", false))).
		WithArgs("User").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL("Group", false))).
		WithArgs("Group").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("User", "04i-1", "005-user", nil).
		WillReturnResult(sqlmock.NewResult(0, 0))

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil {
		t.Fatal("BackfillPolymorphicDiscriminators = nil, want concurrent source change error")
	}
	if !strings.Contains(err.Error(), "concurrent source row change") {
		t.Fatalf("error = %v, want concurrent source change diagnostics", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsBatchesTargetQueries(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	firstPage := sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"})
	firstTargetPage := sqlmock.NewRows([]string{"Id"})
	firstTargetArgs := make([]driver.Value, polymorphicBackfillBatchSize)
	for i := 0; i < polymorphicBackfillBatchSize; i++ {
		rowID := fmt.Sprintf("04i-%03d", i)
		fkID := fmt.Sprintf("005-%03d", i)
		firstPage.AddRow(rowID, fkID, sql.NullString{String: "User", Valid: true})
		firstTargetPage.AddRow(fkID)
		firstTargetArgs[i] = fkID
	}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(firstPage)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", polymorphicBackfillBatchSize))).
		WithArgs(firstTargetArgs...).
		WillReturnRows(firstTargetPage)

	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectAfterSQL())).
		WithArgs("04i-499").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-500", "005-500", sql.NullString{String: "User", Valid: true}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-500").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-500"))

	def := polyBackfillTestDef()
	def.Columns[0].ReferenceTo = []string{"User"}
	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{def}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsChunksCASWritesByTargetType(t *testing.T) {
	const writeChunkSize = 100

	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	rowCount := writeChunkSize + 1
	page := sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"})
	targetPage := sqlmock.NewRows([]string{"Id"})
	targetArgs := make([]driver.Value, rowCount)
	for i := 0; i < rowCount; i++ {
		rowID := fmt.Sprintf("04i-%03d", i)
		fkID := fmt.Sprintf("005-%03d", i)
		page.AddRow(rowID, fkID, sql.NullString{})
		targetPage.AddRow(fkID)
		targetArgs[i] = fkID
	}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(page)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", rowCount))).
		WithArgs(targetArgs...).
		WillReturnRows(targetPage)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("Group", rowCount))).
		WithArgs(targetArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	firstArgs := make([]driver.Value, 0, 1+3*writeChunkSize)
	firstArgs = append(firstArgs, "User")
	for i := 0; i < writeChunkSize; i++ {
		firstArgs = append(firstArgs, fmt.Sprintf("04i-%03d", i), fmt.Sprintf("005-%03d", i), nil)
	}
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", writeChunkSize))).
		WithArgs(firstArgs...).
		WillReturnResult(sqlmock.NewResult(0, writeChunkSize))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("User", "04i-100", "005-100", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	def := polyBackfillTestDef()
	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{def}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsAdvancesPastEmptyPrimaryKeyPage(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	firstPage := sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"})
	for i := 0; i < polymorphicBackfillBatchSize; i++ {
		firstPage.AddRow(fmt.Sprintf("04i-%03d", i), nil, nil)
	}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(firstPage)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectAfterSQL())).
		WithArgs("04i-499").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-500", "005-user", sql.NullString{}))
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))
	mock.ExpectExec(regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))).
		WithArgs("User", "04i-500", "005-user", nil).
		WillReturnResult(sqlmock.NewResult(0, 1))

	def := polyBackfillTestDef()
	def.Columns[0].ReferenceTo = []string{"User"}
	if err := BackfillPolymorphicDiscriminators(context.Background(), db, []domainschema.TableDefinition{def}); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestResolvePolymorphicDiscriminatorTargetsRetriesTimedOutBatchQuery(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	query := regexp.QuoteMeta(polyBackfillTargetSelectSQL("User", 1))
	mock.ExpectQuery(query).
		WithArgs("005-user").
		WillReturnError(context.DeadlineExceeded)
	mock.ExpectQuery(query).
		WithArgs("005-user").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("005-user"))

	targets, err := resolvePolymorphicDiscriminatorTargets(
		context.Background(),
		db,
		[]string{"User"},
		[]polymorphicBackfillRow{{
			id:   "04i-1",
			fkID: sql.NullString{String: "005-user", Valid: true},
		}},
		&polymorphicBackfillResolutionState{unavailableTargetErrors: make(map[string]error)},
	)
	if err != nil {
		t.Fatalf("resolvePolymorphicDiscriminatorTargets: %v", err)
	}
	if got := targets["005-user"]; len(got) != 1 || got[0] != "User" {
		t.Fatalf("targets[005-user] = %v, want [User]", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsRetriesTimedOutRowLoadAndDrain(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	query := regexp.QuoteMeta(polyBackfillSelectSQL())
	mock.ExpectQuery(query).
		WillReturnError(context.DeadlineExceeded)
	mock.ExpectQuery(query).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", "005-user", sql.NullString{}).
			RowError(0, context.DeadlineExceeded))
	mock.ExpectQuery(query).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	if err := BackfillPolymorphicDiscriminators(context.Background(), db, defs); err != nil {
		t.Fatalf("BackfillPolymorphicDiscriminators: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func TestBackfillPolymorphicDiscriminatorsDoesNotRetryTimedOutUpdate(t *testing.T) {
	db, mock, cleanup := newPolyBackfillSQLMock(t)
	defer cleanup()

	defs := []domainschema.TableDefinition{polyBackfillTestDef()}
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillSelectSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}).
			AddRow("04i-1", nil, sql.NullString{String: "User", Valid: true}))
	update := regexp.QuoteMeta(polyBackfillUpdateSQL("ProcessInstanceWorkitem", "ActorId", "ActorId_type", 1))
	mock.ExpectExec(update).
		WithArgs(nil, "04i-1", nil, "User").
		WillReturnError(context.DeadlineExceeded)

	err := BackfillPolymorphicDiscriminators(context.Background(), db, defs)
	if err == nil || !strings.Contains(err.Error(), context.DeadlineExceeded.Error()) {
		t.Fatalf("BackfillPolymorphicDiscriminators error = %v, want deadline exceeded", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

func polyBackfillStaleCandidateSQL(afterID bool) string {
	where := "(source.`ActorId` IS NULL OR BINARY source.`ActorId` = BINARY '') AND source.`ActorId_type` IS NOT NULL AND NOT (BINARY source.`ActorId_type` <=> BINARY '')"
	if afterID {
		where = "source.`Id` > ? AND (" + where + ")"
	}
	return "SELECT /* polymorphic-backfill-stale-discriminator */ source.`Id`, source.`ActorId`, source.`ActorId_type` FROM `ProcessInstanceWorkitem` AS source WHERE " + where + " ORDER BY source.`Id` LIMIT 500"
}

func polyBackfillInvalidSourceIDSQL() string {
	return "SELECT /* polymorphic-backfill-invalid-source-id */ source.`Id`, source.`ActorId`, source.`ActorId_type` FROM `ProcessInstanceWorkitem` AS source WHERE source.`Id` IS NULL OR source.`Id` = '' LIMIT 1"
}

func expectValidPolyBackfillSourceIDs(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillInvalidSourceIDSQL())).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
}

func polyBackfillTargetCandidateSQL(target string, afterID bool) string {
	where := "NOT (BINARY source.`ActorId_type` <=> BINARY ?)"
	if afterID {
		where = "source.`Id` > ? AND (" + where + ")"
	}
	return fmt.Sprintf(
		"SELECT /* polymorphic-backfill-target-candidate */ source.`Id`, source.`ActorId`, source.`ActorId_type` FROM `ProcessInstanceWorkitem` AS source INNER JOIN `%s` AS matched ON matched.`Id` = source.`ActorId` WHERE %s ORDER BY source.`Id` LIMIT 500",
		target, where,
	)
}

func expectEmptyPolyBackfillCandidateProbes(mock sqlmock.Sqlmock) {
	expectValidPolyBackfillSourceIDs(mock)
	mock.ExpectQuery(regexp.QuoteMeta(polyBackfillStaleCandidateSQL(false))).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	for _, target := range []string{"User", "Group"} {
		mock.ExpectQuery(regexp.QuoteMeta(polyBackfillTargetCandidateSQL(target, false))).
			WithArgs(target).
			WillReturnRows(sqlmock.NewRows([]string{"Id", "ActorId", "ActorId_type"}))
	}
}

func polyBackfillCombinedSourceSelectSQL(afterID bool) string {
	query := "SELECT source.`Id`, source.`ActorId`, source.`ActorId_type`, target_0.`Id`, target_1.`Id` FROM (SELECT /* polymorphic-backfill-classification */ `Id`, `ActorId`, `ActorId_type` FROM `ProcessInstanceWorkitem`"
	if afterID {
		query += " WHERE `Id` > ?"
	}
	return query + " ORDER BY `Id` LIMIT 500) AS source LEFT JOIN `User` AS target_0 ON target_0.`Id` = source.`ActorId` LEFT JOIN `Group` AS target_1 ON target_1.`Id` = source.`ActorId` ORDER BY source.`Id`"
}

func polyBackfillSelectSQL() string {
	return "SELECT `Id`, `ActorId`, `ActorId_type` FROM `ProcessInstanceWorkitem` ORDER BY `Id` LIMIT 500"
}

func polyBackfillSelectAfterSQL() string {
	return "SELECT `Id`, `ActorId`, `ActorId_type` FROM `ProcessInstanceWorkitem` WHERE `Id` > ? ORDER BY `Id` LIMIT 500"
}

func groupMemberPolyBackfillSelectSQL() string {
	return "SELECT `Id`, `UserOrGroupId`, `UserOrGroupId_type` FROM `GroupMember` ORDER BY `Id` LIMIT 500"
}

func polyBackfillRelatedToSelectSQL() string {
	return "SELECT `Id`, `RelatedToId`, `RelatedToId_type` FROM `EmailMessage` ORDER BY `Id` LIMIT 500"
}

func polyBackfillTargetSelectSQL(target string, argCount int) string {
	requestedRows := make([]string, argCount)
	for i := range requestedRows {
		if i == 0 {
			requestedRows[i] = "SELECT ? AS `RequestedId`"
		} else {
			requestedRows[i] = "SELECT ?"
		}
	}
	return fmt.Sprintf(
		"SELECT requested.`RequestedId` FROM (%s) AS requested INNER JOIN `%s` AS matched ON matched.`Id` = requested.`RequestedId`",
		strings.Join(requestedRows, " UNION ALL "), target,
	)
}

func polyBackfillObjectKeyPrefixOwnersSQL() string {
	return "SELECT `KeyPrefix`, `ApiName` FROM `_System_Object` WHERE `IsDeleted` = 0 AND `KeyPrefix` IS NOT NULL AND `KeyPrefix` <> '' ORDER BY `KeyPrefix`, `ApiName`"
}

func polyBackfillUpdateSQL(table, fkColumn, typeColumn string, rowCount int) string {
	predicate := fmt.Sprintf("(`Id` = ? AND `%s` <=> ? AND `%s` <=> ?)", fkColumn, typeColumn)
	return fmt.Sprintf("UPDATE `%s` SET `%s` = ? WHERE %s", table, typeColumn, strings.TrimSuffix(strings.Repeat(predicate+" OR ", rowCount), " OR "))
}

func newPolyBackfillSQLMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	return db, mock, func() { _ = db.Close() }
}

func polyBackfillTestDef() domainschema.TableDefinition {
	return domainschema.TableDefinition{
		TableName: "ProcessInstanceWorkitem",
		Columns: []domainschema.ColumnDefinition{{
			Name:          "ActorId",
			Type:          "VARCHAR(255)",
			LogicalType:   string(constants.FieldTypeReference),
			ReferenceTo:   []string{"User", "Group"},
			IsPolymorphic: true,
		}},
	}
}

func polyBackfillRelatedToTestDef() domainschema.TableDefinition {
	return domainschema.TableDefinition{
		TableName: "EmailMessage",
		Columns: []domainschema.ColumnDefinition{{
			Name:          "RelatedToId",
			Type:          "VARCHAR(255)",
			LogicalType:   string(constants.FieldTypeReference),
			ReferenceTo:   []string{"AppointmentCategory", "Opportunity"},
			IsPolymorphic: true,
		}},
	}
}

func findSystemTableForPolyTest(t *testing.T, defs []domainschema.TableDefinition, tableName string) domainschema.TableDefinition {
	t.Helper()
	for _, def := range defs {
		if def.TableName == tableName {
			return def
		}
	}
	t.Fatalf("table %s not found", tableName)
	return domainschema.TableDefinition{}
}