package automation

import (
	"context"
	"errors"
	"sync"
	"testing"
)

// The inverse resolver relies on the existing active-locator unique key rather
// than replacing it with an application-level race check. This DB-backed test
// proves two concurrent owners for the same remote locator cannot both commit.
func TestSQLConnectorPushConcurrentAdoptKeepsOneRemoteLocatorOwner(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	ctx := context.Background()
	createConnectorRuntimeOwnerCatalogTables(t, db)
	for _, ddl := range []string{connectorRemoteLinkTestDDL, connectorPushAttemptTestDDL} {
		if _, err := db.ExecContext(ctx, ddl); err != nil {
			t.Fatalf("create table: %v", err)
		}
	}

	first := connectorPushTestIdentity()
	second := first
	second.ActionSlotID = "other-action-slot"
	second.SourceRecordID = "003-other"
	remoteID := "shared-remote"
	snapshot := connectorPushTestSnapshot()
	adoptions := []ConnectorPushLinkAdoption{
		{RemoteID: remoteID, Identity: first, Snapshot: snapshot},
		{RemoteID: remoteID, Identity: second, Snapshot: snapshot},
	}
	for i := range adoptions {
		adoptions[i].LinkKey = mustConnectorPushTestLinkKey(adoptions[i].Identity)
	}

	start := make(chan struct{})
	errs := make(chan error, len(adoptions))
	var wg sync.WaitGroup
	for _, adoption := range adoptions {
		adoption := adoption
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_, err := NewSQLConnectorPushDurableStore(db).AdoptFinalizedLink(ctx, adoption)
			errs <- err
		}()
	}
	close(start)
	wg.Wait()
	close(errs)

	succeeded, conflicted := 0, 0
	for err := range errs {
		switch {
		case err == nil:
			succeeded++
		case errors.Is(err, ErrConnectorPushDurableConflict):
			conflicted++
		default:
			t.Fatalf("concurrent AdoptFinalizedLink error = %v", err)
		}
	}
	if succeeded != 1 || conflicted != 1 {
		t.Fatalf("concurrent results success/conflict = %d/%d, want 1/1", succeeded, conflicted)
	}
}