package connector_outboundintent

import (
	"database/sql"
	"fmt"
	"strconv"
	"strings"
	"testing"
	"time"
)

func TestRepositoryDependencyBacklogStaysOutOfRunnableQueriesAgainstTiDB(t *testing.T) {
	db, cleanup := openOutboundIntentTestTiDB(t)
	defer cleanup()
	createOutboundIntentTestTables(t, db)

	now := time.Now().UTC().Truncate(time.Microsecond)
	scope := outboundIntentTestScope()
	tx, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}
	statement, err := tx.Prepare(`INSERT INTO _System_IntegrationOutboundCommand
(Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
 IntentId, LinkKey, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
 SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName, FlowActionName,
 ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload, State, DependencyKind,
 DependencyKey, DependencyReadinessKeys, DependencyBlockedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, 'UPSERT', 'SourceObject', ?, 'connector-owner',
        'FlowName', 'PushAction', 'action-slot', 'destination-slot', JSON_ARRAY(), ?,
        ?, ?, ?, JSON_ARRAY(?), ?, ?, ?)`)
	if err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	for index := 0; index < 1000; index++ {
		identity := fmt.Sprintf("%04d", index)
		if _, err := statement.Exec(
			"waiter-"+identity,
			scope.OrgID,
			scope.AppInstallationID,
			scope.AuthorizedWorkloadID,
			scopeDigest(scope),
			scope.CatalogRevision,
			"intent-"+identity,
			"link-"+identity,
			"idempotency-"+identity,
			"sha256:"+strings.Repeat("a", 64),
			"record-"+identity,
			[]byte(`{"version":1}`),
			stateWaitingDependency,
			dependencyKindAssociationRemoteLink,
			"integration_association_dependency_v1:"+strings.Repeat("b", 60)+identity,
			"link-"+identity,
			now,
			now,
			now,
		); err != nil {
			_ = statement.Close()
			_ = tx.Rollback()
			t.Fatalf("insert waiter %d: %v", index, err)
		}
	}
	if err := statement.Close(); err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ANALYZE TABLE _System_IntegrationOutboundCommand`); err != nil {
		t.Fatal(err)
	}

	repository := NewRepository(db)
	repository.now = func() time.Time { return now.Add(time.Hour) }
	started := time.Now()
	if hinted, err := repository.HasWorkHint(t.Context(), scope); err != nil || hinted {
		t.Fatalf("1,000 waiting commands HasWorkHint = %v, %v, want false, nil", hinted, err)
	}
	hintElapsed := time.Since(started)
	started = time.Now()
	if next, scheduled, err := repository.NextWorkAt(t.Context(), scope); err != nil || scheduled {
		t.Fatalf("1,000 waiting commands NextWorkAt = %v, %v, %v, want zero, false, nil", next, scheduled, err)
	}
	scheduleElapsed := time.Since(started)
	started = time.Now()
	if command, claimed, err := repository.ClaimNext(
		t.Context(), scope, "quiet-backlog-worker", "quiet-backlog-token", time.Minute,
	); err != nil || claimed {
		t.Fatalf("1,000 waiting commands ClaimNext = %#v, %v, %v, want empty, false, nil", command, claimed, err)
	}
	claimElapsed := time.Since(started)

	var total, waiting int
	var retries int64
	if err := db.QueryRow(`SELECT COUNT(*), SUM(State = ?), SUM(RetryCount)
FROM _System_IntegrationOutboundCommand`, stateWaitingDependency).Scan(&total, &waiting, &retries); err != nil {
		t.Fatal(err)
	}
	if total != 1000 || waiting != 1000 || retries != 0 {
		t.Fatalf("quiet backlog mutated: total=%d waiting=%d retries=%d", total, waiting, retries)
	}

	digest := scopeDigest(scope)
	plans := []struct {
		name  string
		query string
		args  []any
	}{
		{
			name:  "dependency_reconcile_page",
			query: dependencyReconcileListSQL,
			args: []any{
				"foreign-scope-digest", scope.CatalogRevision, stateWaitingDependency,
				dependencyKindAssociationRemoteLink, "", maxDependencyReconcileBatch + 1,
			},
		},
		{
			name:  "lease_recovery",
			query: outboundLeaseRecoverySQL,
			args: []any{
				StatePending,
				"association_create_outcome_uncertain", "association_update_outcome_uncertain", StateRetry,
				StateUncertain,
				"association_create_outcome_uncertain", "association_update_outcome_uncertain",
				now, digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
				scope.CatalogRevision, StateClaimed, StateDispatching, now,
			},
		},
		{
			name:  "claim_candidate",
			query: outboundClaimCandidateSQL,
			args: []any{
				digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
				scope.CatalogRevision, StatePending, StateRetry, stateDependencyRegistration, now,
				StatePending, StateClaimed, StateDispatching, StateRetry, stateDependencyRegistration,
				stateWaitingDependency, StateWriteback, StateUncertain,
			},
		},
		{
			name:  "next_work_at",
			query: outboundNextWorkAtSQL,
			args: []any{
				now, now,
				digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
				scope.CatalogRevision, StateWriteback,
				now,
				digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
				scope.CatalogRevision, StateClaimed, StateDispatching,
				now,
				digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
				scope.CatalogRevision, StatePending, StateRetry, stateDependencyRegistration,
				StatePending, StateClaimed, StateDispatching, StateRetry, stateDependencyRegistration,
				stateWaitingDependency, StateWriteback, StateUncertain,
			},
		},
	}
	for _, plan := range plans {
		rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+plan.query, plan.args...)
		if err != nil {
			t.Fatalf("explain %s: %v", plan.name, err)
		}
		scanned, fullScan := outboundDependencyBacklogScanRows(t, rows)
		if fullScan || scanned > 8 {
			t.Fatalf("%s scans quiet dependency backlog: rows=%d full_scan=%t", plan.name, scanned, fullScan)
		}
		t.Logf("plan=%s scan_rows=%d full_scan=%t", plan.name, scanned, fullScan)
	}
	t.Logf(
		"1,000 waiting commands: HasWorkHint=%s NextWorkAt=%s ClaimNext=%s",
		hintElapsed, scheduleElapsed, claimElapsed,
	)
}

func outboundDependencyBacklogScanRows(t *testing.T, rows *sql.Rows) (int, bool) {
	t.Helper()
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	values := make([]sql.NullString, len(columns))
	targets := make([]any, len(columns))
	for index := range values {
		targets[index] = &values[index]
	}
	total, fullScan, observed := 0, false, false
	for rows.Next() {
		if err := rows.Scan(targets...); err != nil {
			t.Fatal(err)
		}
		row := make(map[string]string, len(columns))
		for index, column := range columns {
			row[column] = values[index].String
		}
		if strings.Contains(row["id"], "Point_Get") {
			observed = true
		}
		if !strings.Contains(row["id"], "Scan") {
			continue
		}
		observed = true
		count, err := strconv.ParseFloat(row["actRows"], 64)
		if err != nil {
			t.Fatalf("parse %s scan rows %q: %v", row["id"], row["actRows"], err)
		}
		total += int(count)
		fullScan = fullScan || strings.Contains(row["id"], "FullScan")
		t.Logf("scan=%s rows=%s access=%s", row["id"], row["actRows"], row["access object"])
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !observed {
		t.Fatal("plan has no observed data access; folded EXISTS is not zero-scan evidence")
	}
	return total, fullScan
}