package automation

// Inbound finalization must admit exactly one owner for a remote locator,
// including when competing transactions start together on real TiDB.

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"
)

func TestConnectorInboundFinalizationConcurrentRemoteLocatorOwnership(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	db.SetMaxOpenConns(8)
	ctx := context.Background()

	createConnectorRuntimeOwnerCatalogTables(t, db)
	for _, ddl := range []string{connectorRemoteLinkTestDDL, connectorPushAttemptTestDDL, connectorInboundEvidenceTestDDL} {
		if _, err := db.ExecContext(ctx, ddl); err != nil {
			t.Fatalf("create table: %v", err)
		}
	}
	t.Run("ambient_inbound_remote_locator_fan_in_no_retry", func(t *testing.T) {
		const iterations = 20
		for i := 0; i < iterations; i++ {
			left := colocationIdentity(fmt.Sprintf("ambient-fanin-left-%d", i))
			right := colocationIdentity(fmt.Sprintf("ambient-fanin-right-%d", i))
			left.DestinationSlotID = "shared-ambient-destination"
			right.DestinationSlotID = left.DestinationSlotID
			remoteID := fmt.Sprintf("shared-ambient-remote-%d", i)
			var leftErr, rightErr error
			start := make(chan struct{})
			var wg sync.WaitGroup
			wg.Add(2)
			go func() {
				defer wg.Done()
				<-start
				leftErr = colocationFinalizeInbound(ctx, db, left, remoteID)
			}()
			go func() {
				defer wg.Done()
				<-start
				rightErr = colocationFinalizeInbound(ctx, db, right, remoteID)
			}()
			close(start)
			wg.Wait()

			if !((leftErr == nil && errors.Is(rightErr, ErrConnectorPushDurableConflict)) ||
				(rightErr == nil && errors.Is(leftErr, ErrConnectorPushDurableConflict))) {
				t.Fatalf("iter %d: want exactly one fan-in winner; left=%v right=%v", i, leftErr, rightErr)
			}
			locator := ConnectorPushRemoteLocatorHash(left, remoteID)
			var links int
			if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE RemoteLocatorActiveKey = ?", locator).Scan(&links); err != nil {
				t.Fatalf("iter %d: count locator links: %v", i, err)
			}
			if links != 1 {
				t.Fatalf("iter %d: active locator links=%d, want 1", i, links)
			}
		}
	})
}

func colocationIdentity(recordID string) ConnectorPushDurableIdentity {
	identity := connectorPushTestIdentity()
	identity.SourceRecordID = recordID
	return identity
}

func colocationFinalizeInbound(ctx context.Context, db *sql.DB, identity ConnectorPushDurableIdentity, remoteID string) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	snapshot := connectorPushTestSnapshot()
	snapshot.ConnectorNamespace = "connector-owner-key"
	snapshot.RemoteIDField = "ExternalId__c"
	snapshot.ProfileContractHash = ConnectorPushProfileContractHash(snapshot)
	finalization := ConnectorInboundRemoteIdentityFinalization{
		Evidence: ConnectorInboundRemoteIDEvidence{
			ID:                 "ev-" + identity.SourceRecordID,
			ConnectorOwnerKey:  identity.ConnectorOwnerKey,
			ConnectorNamespace: "connector-routing-namespace",
			SourceObject:       identity.SourceObject,
			SourceRecordId:     identity.SourceRecordID,
			RemoteIDField:      snapshot.RemoteIDField,
			RemoteId:           remoteID,
		},
		Adoption: ConnectorPushLinkAdoption{
			RemoteID: remoteID,
			Identity: identity,
			Snapshot: snapshot,
		},
	}
	store := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx})
	if err := store.LockConnectorRemoteIdentityLaneInAmbientTx(ctx, identity, remoteID); err != nil {
		return err
	}
	if _, err := store.CheckConnectorRemoteIdentityRegistrationInAmbientTx(ctx, identity, remoteID); err != nil {
		return err
	}
	if err := store.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil {
		return err
	}
	if err := store.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil {
		return err
	}
	return tx.Commit()
}

func colocationWaitForSignal(t *testing.T, ctx context.Context, signal <-chan struct{}, failure string) {
	t.Helper()
	select {
	case <-signal:
	case <-ctx.Done():
		t.Fatalf("%s: %v", failure, ctx.Err())
	}
}

func colocationAssertStillBlocked(t *testing.T, done <-chan error, failure string) {
	t.Helper()
	select {
	case err := <-done:
		t.Fatalf("%s: %v", failure, err)
	case <-time.After(150 * time.Millisecond):
	}
}