package connector_outboundintent

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	"github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture"
)

func TestRemoteLinkDependencyProbesAreExactPointReadsTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_ConnectorRemoteLink")
	scope := outboundIntentTestScope()
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback()
	var presentKey string
	for i := 0; i < 1000; i++ {
		intent := outboundIntentTestIntent(scope.CatalogRevision, fmt.Sprintf("intent-%04d", i), fmt.Sprintf("record-%04d", i))
		producer := Command{Scope: scope, Intent: intent, LinkKey: outboundLinkKey(scope, intent)}
		finalization := outboundIntentTestFinalization(producer, fmt.Sprintf("remote-%04d", i))
		if err := insertRemoteLink(t.Context(), tx, finalization); err != nil {
			t.Fatal(err)
		}
		presentKey = producer.LinkKey
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ANALYZE TABLE _System_ConnectorRemoteLink`); err != nil {
		t.Fatal(err)
	}
	for _, tc := range []struct {
		name, key string
		want      bool
	}{
		{"absent", "absent-link-key", false},
		{"present", presentKey, true},
		{"wrong_case", strings.ToUpper(presentKey), false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ready, err := remoteLinkDependencyPresent(t.Context(), db, []string{tc.key})
			if err != nil || ready != tc.want {
				t.Fatalf("safety ready=%v err=%v", ready, err)
			}
			tx, cleanup, err := sqltx.BeginPessimisticTx(t.Context(), db)
			if err != nil {
				t.Fatal(err)
			}
			defer cleanup()
			ready, err = remoteLinkDependencyReady(t.Context(), tx, []string{tc.key})
			if err != nil || ready != tc.want {
				t.Fatalf("registration ready=%v err=%v", ready, err)
			}
			if err := tx.Rollback(); err != nil {
				t.Fatal(err)
			}
			for _, suffix := range []string{"", " FOR UPDATE"} {
				rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+remoteLinkDependencyLookupSQL+suffix, tc.key, tc.key)
				if err != nil {
					t.Fatal(err)
				}
				assertDependencyPointReadPlan(t, rows)
			}
		})
	}
	// A current-slot miss must still allow the declared previous slot.
	if ready, err := remoteLinkDependencyPresent(t.Context(), db, []string{"absent-link-key", presentKey}); err != nil || !ready {
		t.Fatalf("previous-slot readiness=%v err=%v", ready, err)
	}
}

func assertDependencyPointReadPlan(t *testing.T, rows *sql.Rows) {
	t.Helper()
	defer rows.Close()
	cols, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	values := make([]sql.NullString, len(cols))
	targets := make([]any, len(cols))
	for i := range values {
		targets[i] = &values[i]
	}
	point := false
	for rows.Next() {
		if err := rows.Scan(targets...); err != nil {
			t.Fatal(err)
		}
		id := values[0].String
		if strings.Contains(id, "Scan") {
			t.Fatalf("dependency key probe scans instead of point-reading: %s", id)
		}
		point = point || strings.Contains(id, "Point_Get")
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !point {
		t.Fatal("no point-read operator observed; a folded EXISTS plan is not evidence")
	}
}

func TestDependencySafetyBudgetPreservesPartialProgressTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
	scope := outboundIntentTestScope()
	repo := NewRepository(db)
	producerIntent := outboundIntentTestIntent(scope.CatalogRevision, "producer", "source")
	producer := Command{Scope: scope, Intent: producerIntent, LinkKey: outboundLinkKey(scope, producerIntent)}
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := insertRemoteLink(t.Context(), tx, outboundIntentTestFinalization(producer, "remote")); err != nil {
		t.Fatal(err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	var ids []string
	for i := 0; i < 3; i++ {
		command, _, err := repo.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, fmt.Sprintf("wait-%d", i), fmt.Sprintf("record-%d", i)))
		if err != nil {
			t.Fatal(err)
		}
		ids = append(ids, command.CommandID)
	}
	sort.Strings(ids)
	dependency := associationRemoteLinkDependency(scope, producerIntent.ConnectorOwnerKey, producerIntent.SourceObject, producerIntent.SourceRecordID, producerIntent.DestinationSlotID, []string{producer.LinkKey})
	for _, id := range ids {
		if _, err := db.Exec(`UPDATE _System_IntegrationOutboundCommand
SET State=?, DependencyKind=?, DependencyKey=?, DependencyReadinessKeys=JSON_ARRAY(?), DependencyBlockedAt=UTC_TIMESTAMP(6)
WHERE Id=?`, stateWaitingDependency, dependency.Kind, dependency.Key, producer.LinkKey, id); err != nil {
			t.Fatal(err)
		}
	}
	// Hold the second CAS in another real pessimistic transaction. The first
	// wake must survive the deadline, and the next turn must revisit the second.
	lock, cleanup, err := sqltx.BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer cleanup()
	var lockedID string
	if err := lock.QueryRowContext(t.Context(), `SELECT Id FROM _System_IntegrationOutboundCommand WHERE Id=? FOR UPDATE`, ids[1]).Scan(&lockedID); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 300*time.Millisecond)
	cursor, more, woken, _, err := repo.reconcileDependencyWaits(ctx, scope, "", 3)
	cancel()
	if !errors.Is(err, context.DeadlineExceeded) || cursor != ids[0] || !more || woken != 1 {
		t.Fatalf("interrupted page cursor=%q more=%v woken=%d err=%v", cursor, more, woken, err)
	}
	if err := lock.Rollback(); err != nil {
		t.Fatal(err)
	}
	cursor, more, woken, _, err = repo.reconcileDependencyWaits(t.Context(), scope, cursor, 1)
	if err != nil || cursor != ids[1] || !more || woken != 1 {
		t.Fatalf("resumed bounded page cursor=%q more=%v woken=%d err=%v", cursor, more, woken, err)
	}
	cursor, more, woken, _, err = repo.reconcileDependencyWaits(t.Context(), scope, cursor, 1)
	if err != nil || cursor != "" || more || woken != 1 {
		t.Fatalf("final page cursor=%q more=%v woken=%d err=%v", cursor, more, woken, err)
	}
	var ready, retries, attempts int
	if err := db.QueryRow(`SELECT SUM(State='retry'), SUM(RetryCount), SUM(AttemptCount) FROM _System_IntegrationOutboundCommand`).Scan(&ready, &retries, &attempts); err != nil {
		t.Fatal(err)
	}
	if ready != 3 || retries != 0 || attempts != 0 {
		t.Fatalf("ready=%d retries=%d attempts=%d", ready, retries, attempts)
	}
}