//go:build integration
// +build integration

package integration_test

import (
	"context"
	"database/sql"
	"errors"
	"flag"
	"fmt"
	"math/rand"
	"net"
	"net/http"
	"net/http/pprof"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"
	_ "github.com/go-sql-driver/mysql"
	"go.uber.org/zap"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/executor"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

type mysqlConfig struct {
	host     string
	port     int
	user     string
	password string
}

type sessionOptions struct {
	protoVersion       int
	keyspace           string
	disableHostLookup  bool
	numConns           int
	connectTimeout     time.Duration
	queryTimeout       time.Duration
	allowedAuthClasses []string
	username           string
	password           string
	compressor         gocql.Compressor
	skipAuth           bool
}

var (
	mysqlCfg        mysqlConfig
	mysqlConfigured bool
	proxyHost       string
	proxyPort       int
	proxyStop       func()
	startOnce       sync.Once
	startErr        error
)

func TestMain(m *testing.M) {
	flag.Parse()
	cfg, err := loadMySQLConfig()
	if err != nil {
		fmt.Fprintf(os.Stderr, "integration: %v\n", err)
		os.Exit(1)
	}
	mysqlCfg = cfg
	mysqlConfigured = true
	rand.Seed(time.Now().UnixNano())

	if err := checkMySQL(mysqlCfg); err != nil {
		fmt.Fprintf(os.Stderr, "integration: MySQL check failed: %v\n", err)
		os.Exit(1)
	}
	startOnce.Do(func() {
		proxyHost, proxyPort, proxyStop, startErr = startProxy(mysqlCfg)
	})
	if startErr != nil {
		fmt.Fprintf(os.Stderr, "integration: proxy start failed: %v\n", startErr)
		os.Exit(1)
	}

	code := m.Run()
	if proxyStop != nil {
		proxyStop()
	}
	os.Exit(code)
}

func requireProxy(t *testing.T) {
	t.Helper()
	requireMySQLConfig(t)
	if startErr != nil {
		t.Fatalf("integration: proxy start failed: %v", startErr)
	}
}

func requireMySQLConfig(t *testing.T) {
	t.Helper()
	if !mysqlConfigured {
		t.Fatalf("integration: required MySQL env vars not set; set MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD")
	}
}

func loadMySQLConfig() (mysqlConfig, error) {
	host, hostSet := os.LookupEnv("MYSQL_HOST")
	portStr, portSet := os.LookupEnv("MYSQL_PORT")
	user, userSet := os.LookupEnv("MYSQL_USER")
	password, passwordSet := os.LookupEnv("MYSQL_PASSWORD")

	host = strings.TrimSpace(host)
	portStr = strings.TrimSpace(portStr)
	user = strings.TrimSpace(user)

	missing := make([]string, 0, 4)
	if !hostSet || host == "" {
		missing = append(missing, "MYSQL_HOST")
	}
	if !portSet || portStr == "" {
		missing = append(missing, "MYSQL_PORT")
	}
	if !userSet || user == "" {
		missing = append(missing, "MYSQL_USER")
	}
	if !passwordSet {
		missing = append(missing, "MYSQL_PASSWORD")
	}
	if len(missing) > 0 {
		return mysqlConfig{}, fmt.Errorf("required MySQL env vars not set: %s", strings.Join(missing, ", "))
	}

	port, err := strconv.Atoi(portStr)
	if err != nil {
		return mysqlConfig{}, fmt.Errorf("invalid MYSQL_PORT %q: %w", portStr, err)
	}

	return mysqlConfig{
		host:     host,
		port:     port,
		user:     user,
		password: password,
	}, nil
}

func checkMySQL(cfg mysqlConfig) error {
	dsn := mysqlDSN(cfg)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return err
	}
	defer db.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	return db.PingContext(ctx)
}

func mysqlDSN(cfg mysqlConfig) string {
	params := url.Values{}
	params.Add("parseTime", "true")
	params.Add("loc", "UTC")
	return fmt.Sprintf("%s:%s@tcp(%s:%d)/?%s",
		url.QueryEscape(cfg.user),
		url.QueryEscape(cfg.password),
		cfg.host,
		cfg.port,
		params.Encode(),
	)
}

func startProxy(cfg mysqlConfig) (string, int, func(), error) {
	return startProxyWithConfig(cfg, nil)
}

func startProxyWithConfig(cfg mysqlConfig, apply func(*config.Config)) (string, int, func(), error) {
	host, port, stop, _, err := startProxyWithConfigAndPool(cfg, apply)
	return host, port, stop, err
}

func startProxyWithConfigAndPool(cfg mysqlConfig, apply func(*config.Config)) (string, int, func(), *mysqlbackend.PoolManager, error) {
	return startProxyWithConfigAndPoolLogger(cfg, apply, zap.NewNop())
}

func startProxyWithConfigAndPoolLogger(cfg mysqlConfig, apply func(*config.Config), log *zap.Logger) (string, int, func(), *mysqlbackend.PoolManager, error) {
	if log == nil {
		log = zap.NewNop()
	}
	conf := config.DefaultConfig()
	conf.MySQL.Host = fmt.Sprintf("%s:%d", cfg.host, cfg.port)
	conf.MySQL.MaxOpenConns = 10
	conf.MySQL.MaxIdleConns = 5
	conf.MySQL.ConnectTimeout = 5 * time.Second
	conf.MySQL.ReadTimeout = 5 * time.Second
	conf.MySQL.WriteTimeout = 5 * time.Second
	conf.MySQL.ConnMaxLifetime = 2 * time.Minute
	conf.MySQL.ConnMaxIdleTime = 30 * time.Second
	conf.Server.Listeners = []config.ServerListenerConfig{
		{Name: "plain", Address: "127.0.0.1:0", TLS: config.ServerTLSConfig{Mode: "disabled"}},
	}
	conf.Server.MaxConnections = 100
	if apply != nil {
		apply(conf)
	}
	config.SetConfig(conf)

	metrics.Init(metrics.Settings{
		Enabled:              conf.Server.Admin.Metrics.Enabled,
		SlowRequestThreshold: conf.Server.Admin.Metrics.SlowRequestThreshold,
	}, nil)

	poolCfg := &mysqlbackend.PoolConfig{
		Host:            conf.MySQL.Host,
		MaxOpenConns:    conf.MySQL.MaxOpenConns,
		MaxIdleConns:    conf.MySQL.MaxIdleConns,
		ConnectTimeout:  conf.MySQL.ConnectTimeout,
		ReadTimeout:     conf.MySQL.ReadTimeout,
		WriteTimeout:    conf.MySQL.WriteTimeout,
		ConnMaxLifetime: conf.MySQL.ConnMaxLifetime,
		ConnMaxIdleTime: conf.MySQL.ConnMaxIdleTime,
		DSNParams:       conf.MySQL.DSNParams,
		PreparedStatements: mysqlbackend.PreparedStatementsConfig{
			Enabled:          conf.MySQL.Prepared.Enabled,
			MaxStatements:    conf.MySQL.Prepared.MaxStatements,
			IdleTTL:          conf.MySQL.Prepared.IdleTTL,
			CleanupInterval:  conf.MySQL.Prepared.CleanupInterval,
			PrewarmOnPrepare: conf.MySQL.Prepared.PrewarmOnPrepare,
		},
		TLS: mysqlbackend.TLSConfig{
			Mode:               conf.MySQL.TLS.Mode,
			CAFile:             conf.MySQL.TLS.CAFile,
			CertFile:           conf.MySQL.TLS.CertFile,
			KeyFile:            conf.MySQL.TLS.KeyFile,
			ServerName:         conf.MySQL.TLS.ServerName,
			InsecureSkipVerify: conf.MySQL.TLS.InsecureSkipVerify,
		},
	}

	poolMgr := mysqlbackend.NewPoolManager(poolCfg, log)
	queryExecutor := executor.NewExecutor(poolMgr, log)
	queryExecutor.Start()
	authenticator := cql.NewPasswordAuthenticator()
	handler := cql.NewMessageHandlerWithExecutor(log, authenticator, queryExecutor)

	server := cql.NewServer(conf.Server.Listeners[0].Address, handler, log, &cql.ServerOptions{
		ProtocolVersion:   cql.ProtoVersion4,
		MaxConnections:    conf.Server.MaxConnections,
		MaxInFlightFrames: conf.Server.MaxInFlightFrames,
		FrameInfo:         conf.Logging.FrameInfo,
		Listeners: []cql.ListenerConfig{
			{
				Name: conf.Server.Listeners[0].Name,
				Addr: conf.Server.Listeners[0].Address,
				TLS: &cql.TLSConfig{
					Mode:         conf.Server.Listeners[0].TLS.Mode,
					CertFile:     conf.Server.Listeners[0].TLS.CertFile,
					KeyFile:      conf.Server.Listeners[0].TLS.KeyFile,
					ClientCAFile: conf.Server.Listeners[0].TLS.ClientCAFile,
					ClientAuth:   conf.Server.Listeners[0].TLS.ClientAuth,
					MinVersion:   conf.Server.Listeners[0].TLS.MinVersion,
					MaxVersion:   conf.Server.Listeners[0].TLS.MaxVersion,
				},
			},
		},
	})
	if err := server.Start(); err != nil {
		_ = poolMgr.Close()
		return "", 0, nil, nil, err
	}

	var adminServer *http.Server
	if conf.Server.Admin.Address != "" {
		mux := http.NewServeMux()
		if conf.Server.Admin.PprofEnabled() {
			pprofBase := conf.Server.Admin.PprofPath()
			mux.HandleFunc(pprofBase+"/", pprof.Index)
			mux.HandleFunc(pprofBase+"/cmdline", pprof.Cmdline)
			mux.HandleFunc(pprofBase+"/profile", pprof.Profile)
			mux.HandleFunc(pprofBase+"/symbol", pprof.Symbol)
			mux.HandleFunc(pprofBase+"/trace", pprof.Trace)
		}
		if conf.Server.Admin.Metrics.Enabled {
			path := conf.Server.Admin.Metrics.Path
			if path == "" {
				path = "/metrics"
			}
			if handler := metrics.Handler(); handler != nil {
				mux.Handle(path, handler)
			}
		}
		adminServer = &http.Server{Handler: mux}
		adminLn, err := net.Listen("tcp", conf.Server.Admin.Address)
		if err != nil {
			_ = server.Stop()
			queryExecutor.Close()
			_ = poolMgr.Close()
			return "", 0, nil, nil, err
		}
		go func() {
			_ = adminServer.Serve(adminLn)
		}()
	}

	host, port := parseTCPAddr(server.Addr())
	stop := func() {
		_ = server.Stop()
		if adminServer != nil {
			_ = adminServer.Close()
		}
		queryExecutor.Close()
		_ = poolMgr.Close()
	}
	return host, port, stop, poolMgr, nil
}

func parseTCPAddr(addr net.Addr) (string, int) {
	if addr == nil {
		return "127.0.0.1", 0
	}
	tcpAddr, ok := addr.(*net.TCPAddr)
	if !ok || tcpAddr == nil {
		return "127.0.0.1", 0
	}
	host := "127.0.0.1"
	if tcpAddr.IP != nil && !tcpAddr.IP.IsUnspecified() {
		host = tcpAddr.IP.String()
	}
	return host, tcpAddr.Port
}

func newSession(t *testing.T, opts sessionOptions) *gocql.Session {
	session, err := tryNewSession(t, opts)
	if err != nil {
		t.Fatalf("create session failed: %v", err)
	}
	return session
}

func tryNewSession(t *testing.T, opts sessionOptions) (*gocql.Session, error) {
	t.Helper()
	requireProxy(t)

	if opts.protoVersion == 0 {
		opts.protoVersion = 4
	}
	if opts.numConns == 0 {
		opts.numConns = 1
	}
	if opts.connectTimeout == 0 {
		opts.connectTimeout = 5 * time.Second
	}
	if opts.queryTimeout == 0 {
		opts.queryTimeout = 5 * time.Second
	}

	cluster := gocql.NewCluster(proxyHost)
	cluster.Port = proxyPort
	cluster.ProtoVersion = opts.protoVersion
	cluster.NumConns = opts.numConns
	cluster.ConnectTimeout = opts.connectTimeout
	cluster.Timeout = opts.queryTimeout
	cluster.Keyspace = opts.keyspace
	cluster.DisableInitialHostLookup = opts.disableHostLookup
	if opts.compressor != nil {
		cluster.Compressor = opts.compressor
	}

	if !opts.skipAuth {
		username := opts.username
		if username == "" {
			username = mysqlCfg.user
		}
		password := opts.password
		if password == "" {
			password = mysqlCfg.password
		}

		allowed := opts.allowedAuthClasses
		if len(allowed) == 0 {
			allowed = []string{"org.apache.cassandra.auth.PasswordAuthenticator"}
		}
		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username:              username,
			Password:              password,
			AllowedAuthenticators: allowed,
		}
	}

	session, err := cluster.CreateSession()
	if err != nil {
		return nil, err
	}
	t.Cleanup(func() { session.Close() })
	return session, nil
}

func uniqueKeyspace() string {
	now := time.Now().UnixNano()
	return fmt.Sprintf("it_%x_%x", now, rand.Int31())
}

func createKeyspaceCQL(name string) string {
	return fmt.Sprintf(
		"CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}",
		name,
	)
}

func createTableCQL(keyspace, table string) string {
	return fmt.Sprintf(
		"CREATE TABLE %s.%s (id int, name text, idx int, PRIMARY KEY (id))",
		keyspace,
		table,
	)
}

func createIndexCQL(indexName, keyspace, table, column string) string {
	return fmt.Sprintf(
		"CREATE INDEX %s ON %s.%s (%s)",
		indexName,
		keyspace,
		table,
		column,
	)
}

func dropKeyspaceCQL(name string) string {
	return fmt.Sprintf("DROP KEYSPACE IF EXISTS %s", name)
}

func requireErrorCode(t *testing.T, err error, code int) {
	t.Helper()
	if err == nil {
		t.Fatalf("expected error code %d, got nil", code)
	}
	var reqErr gocql.RequestError
	if !errors.As(err, &reqErr) {
		t.Fatalf("expected RequestError, got %T: %v", err, err)
	}
	if reqErr.Code() != code {
		t.Fatalf("expected error code %d, got %d (%v)", code, reqErr.Code(), err)
	}
}

func waitFor(t *testing.T, desc string, timeout time.Duration, fn func() (bool, error)) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	var lastErr error
	for time.Now().Before(deadline) {
		ok, err := fn()
		if err == nil && ok {
			return
		}
		if err != nil {
			lastErr = err
		}
		time.Sleep(100 * time.Millisecond)
	}
	if lastErr != nil {
		t.Fatalf("timeout waiting for %s: last error: %v", desc, lastErr)
	}
	t.Fatalf("timeout waiting for %s", desc)
}

func scanRows(session *gocql.Session, query string) ([]map[string]interface{}, error) {
	iter := session.Query(query).Iter()
	rows := make([]map[string]interface{}, 0)
	row := map[string]interface{}{}
	for iter.MapScan(row) {
		rows = append(rows, row)
		row = map[string]interface{}{}
	}
	if err := iter.Close(); err != nil {
		return nil, err
	}
	return rows, nil
}

func asString(value interface{}) string {
	switch v := value.(type) {
	case string:
		return v
	case []byte:
		return string(v)
	default:
		return fmt.Sprintf("%v", value)
	}
}