package main

import (
	"context"
	"database/sql"
	"fmt"
	"net"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

// Exercise the tool's real schema and producer together: a stale fixture can
// break admission even when repository tests with a complete schema pass.
func TestNormalEnqueueScenariosTiDB(t *testing.T) {
	for _, tc := range []struct{ name, workType string }{
		{"normal_enqueue_worker_path_800_events_120_lanes", connector_workqueue.WorkTypeConnectorEvent},
		{"workflow_action_worker_path_800_events_120_lanes", connector_workqueue.WorkTypeWorkflowActionApply},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
			defer cancel()
			db := openPressureTestDatabase(t, ctx)
			scenario, ok := findScenario(tc.name)
			if !ok {
				t.Fatal("supported scenario missing")
			}
			// Two events per lane reach batch reconciliation without a load test.
			scenario.Events, scenario.Lanes = 4, 2
			conn, err := db.Conn(ctx)
			if err != nil {
				t.Fatal(err)
			}
			err = createSchema(ctx, conn, scenario)
			conn.Close()
			if err != nil {
				t.Fatal(err)
			}
			if err := seedNormalEnqueue(ctx, db, scenario, time.Now().UTC().Add(-time.Minute)); err != nil {
				t.Fatalf("normal admission: %v", err)
			}
			var events int
			if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM _System_ConnectorEvent").Scan(&events); err != nil || events != 4 {
				t.Fatalf("durable events=%d, err=%v; want 4", events, err)
			}
			rows, err := db.QueryContext(ctx, `SELECT SourceLaneKey, WorkType, COUNT(*),
 SUM(State = 'pending'), SUM(State = 'blocked')
 FROM _System_ConnectorWorkItem GROUP BY SourceLaneKey, WorkType`)
			if err != nil {
				t.Fatal(err)
			}
			defer rows.Close()
			lanes := 0
			for rows.Next() {
				var lane, workType string
				var count, ready, blocked int
				if err := rows.Scan(&lane, &workType, &count, &ready, &blocked); err != nil {
					t.Fatal(err)
				}
				if lane == "" || workType != tc.workType || count != 2 || ready != 1 || blocked != 1 {
					t.Fatalf("lane=%q type=%q count=%d ready=%d blocked=%d", lane, workType, count, ready, blocked)
				}
				lanes++
			}
			if err := rows.Err(); err != nil || lanes != 2 {
				t.Fatalf("lanes=%d, err=%v; want 2", lanes, err)
			}
		})
	}
}

func openPressureTestDatabase(t *testing.T, ctx context.Context) *sql.DB {
	t.Helper()
	config := mysql.NewConfig()
	config.User, config.Passwd = os.Getenv("TIDB_USER"), os.Getenv("TIDB_PASSWORD")
	if config.User == "" {
		config.User = "root"
	}
	host, port := os.Getenv("TIDB_HOST"), os.Getenv("TIDB_PORT")
	if host == "" {
		host = "127.0.0.1"
	}
	if port == "" {
		port = "4001"
	}
	config.Net, config.Addr = "tcp", net.JoinHostPort(host, port)
	config.ParseTime, config.Timeout = true, 2*time.Second
	switch strings.ToLower(strings.TrimSpace(os.Getenv("TIDB_TLS"))) {
	case "1", "true", "yes", "require", "verify-full":
		config.TLSConfig = "true"
	}
	admin, err := sql.Open("mysql", config.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { admin.Close() })
	if err := admin.PingContext(ctx); err != nil {
		if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
			t.Fatalf("required TiDB unavailable: %v", err)
		}
		t.Skipf("local TiDB unavailable: %v", err)
	}
	name := fmt.Sprintf("nexuscrm_pressure_test_%d", time.Now().UnixNano())
	if err := createSeedDatabase(ctx, admin, name); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if _, err := admin.Exec("DROP DATABASE `" + name + "`"); err != nil {
			t.Errorf("drop owned test database: %v", err)
		}
	})
	db, err := openSeedDatabase(ctx, config.FormatDSN(), name)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { db.Close() })
	return db
}