package cluster

import (
	"context"
	"database/sql"
	"errors"
	"linkport/internal/model"
	"linkport/internal/protocol/amqpprobe"
	"linkport/internal/protocol/transport"
	"net"
	"strconv"
	"strings"
	"time"

	"github.com/go-mysql-org/go-mysql/client"
	"github.com/jackc/pgx/v5/pgconn"
	mssql "github.com/microsoft/go-mssqldb"
	"github.com/microsoft/go-mssqldb/msdsn"
	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	"go.mongodb.org/mongo-driver/v2/mongo/options"
)

func probeMongo(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	tls, err := transport.TLSConfig(p)
	if err != nil {
		return nil, err
	}
	authDB := p.AuthSource
	if authDB == "" {
		authDB = "admin"
	}
	opts := options.Client().SetHosts([]string{node(p).Address()}).SetDirect(true).SetDialer(contextDialer{dial}).SetTLSConfig(tls).SetServerSelectionTimeout(3 * time.Second).SetConnectTimeout(3 * time.Second)
	if p.TargetUser != "" {
		opts.SetAuth(options.Credential{AuthSource: authDB, Username: p.TargetUser, Password: p.TargetPassword})
	}
	c, err := mongo.Connect(opts)
	if err != nil {
		return nil, err
	}
	defer func() {
		closeCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		c.Disconnect(closeCtx)
	}()
	var hello bson.M
	err = c.Database("admin").RunCommand(ctx, bson.D{{Key: "hello", Value: 1}}).Decode(&hello)
	if mongoHelloUnsupported(err) {
		hello = nil
		err = c.Database("admin").RunCommand(ctx, bson.D{{Key: "isMaster", Value: 1}}).Decode(&hello)
	}
	if err != nil {
		return nil, err
	}
	info := singleInfo(p, "standalone", "standalone")
	if hello["msg"] == "isdbgrid" {
		info.Type = "mongodb-sharded"
		info.Selected.Role = "router"
		info.Nodes[0].Role = "router"
		return info, nil
	}
	if name, ok := hello["setName"].(string); ok {
		info.Type, info.Name = "mongodb-replica-set", name
		info.Selected.Role = "replica"
		if hello["isWritablePrimary"] == true || hello["ismaster"] == true {
			info.Selected.Role = "primary"
		}
		info.Nodes[0].Role = info.Selected.Role
		for _, key := range []string{"hosts", "passives"} {
			if hosts, ok := hello[key].(bson.A); ok {
				for _, host := range hosts {
					if address, ok := host.(string); ok {
						if n, ok := endpoint(address, 27017); ok {
							if address == hello["primary"] {
								n.Role = "primary"
							}
							info.Nodes = addNodes(info.Nodes, n)
						}
					}
				}
			}
		}
	}
	return info, nil
}

func mongoHelloUnsupported(err error) bool {
	var command mongo.CommandError
	return errors.As(err, &command) && (command.Code == 59 || command.Name == "CommandNotFound")
}

// go-mysql GetString aliases pooled result bytes. Topology outlives Result.Close
// and subsequent queries, so retain owned strings rather than borrowed views.
func mysqlTopologyNode(host string, port int, role string) model.ClusterNode {
	return model.ClusterNode{Host: strings.Clone(host), Port: port, Role: strings.ToLower(strings.Clone(role))}
}

func probeMySQL(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	tls, err := transport.TLSConfig(p)
	if err != nil {
		return nil, err
	}
	var raw net.Conn
	wrapped := func(ctx context.Context, network, address string) (net.Conn, error) {
		c, e := dial(ctx, network, address)
		if e == nil {
			raw = c
			deadline := time.Now().Add(3 * time.Second)
			if d, ok := ctx.Deadline(); ok {
				deadline = d
			}
			c.SetDeadline(deadline)
		}
		return c, e
	}
	c, err := client.ConnectWithDialer(ctx, "tcp", node(p).Address(), p.TargetUser, p.TargetPassword, "", wrapped, func(c *client.Conn) error { c.SetTLSConfig(tls); return nil })
	if err != nil {
		if raw != nil {
			raw.Close()
		}
		return nil, err
	}
	defer c.Close()
	stop := context.AfterFunc(ctx, func() { c.Close() })
	defer stop()
	info := singleInfo(p, "unknown", "unknown")
	if r, e := c.Execute("SELECT @@global.read_only"); e == nil {
		readonly, _ := r.GetInt(0, 0)
		info.Selected.Role = "primary"
		if readonly != 0 {
			info.Selected.Role = "replica"
		}
		info.Nodes[0].Role = info.Selected.Role
		r.Close()
	}
	if r, e := c.Execute("SELECT MEMBER_HOST, MEMBER_PORT, MEMBER_ROLE, MEMBER_STATE FROM performance_schema.replication_group_members"); e == nil {
		if r.RowNumber() > 0 {
			info.Type = "mysql-group-replication"
		}
		for row := 0; row < r.RowNumber(); row++ {
			host, _ := r.GetString(row, 0)
			port, _ := r.GetInt(row, 1)
			role, _ := r.GetString(row, 2)
			state, _ := r.GetString(row, 3)
			if state == "ONLINE" {
				info.Nodes = addNodes(info.Nodes, mysqlTopologyNode(host, int(port), role))
			}
		}
		r.Close()
		if info.Type != "unknown" {
			return info, nil
		}
	}
	r, e := c.Execute("SHOW REPLICA STATUS")
	if e != nil {
		r, e = c.Execute("SHOW SLAVE STATUS")
	}
	if e == nil {
		if r.RowNumber() > 0 {
			info.Type = "mysql-replication"
			host, _ := r.GetStringByName(0, "Source_Host")
			port, _ := r.GetIntByName(0, "Source_Port")
			if host == "" {
				host, _ = r.GetStringByName(0, "Master_Host")
				port, _ = r.GetIntByName(0, "Master_Port")
			}
			info.Nodes = addNodes(info.Nodes, mysqlTopologyNode(host, int(port), "primary"))
		}
		r.Close()
	}
	if info.Type == "unknown" {
		if r, e := c.Execute("SHOW REPLICAS"); e == nil {
			if r.RowNumber() > 0 {
				info.Type = "mysql-replication"
			}
			r.Close()
		}
	}
	if info.Type == "unknown" {
		info.Warning = "可确定节点只读状态，但无法完整读取复制拓扑；单机与暂无副本的主节点无法可靠区分。"
	}
	return info, nil
}

func probePostgres(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	cfg, err := pgconn.ParseConfig("postgres://localhost/postgres?sslmode=disable")
	if err != nil {
		return nil, err
	}
	cfg.Host, cfg.Port, cfg.User, cfg.Password = p.TargetHost, uint16(p.TargetPort), p.TargetUser, p.TargetPassword
	if p.AuthSource != "" && p.AuthSource != "admin" {
		cfg.Database = p.AuthSource
	}
	cfg.TLSConfig, err = transport.TLSConfig(p)
	if err != nil {
		return nil, err
	}
	cfg.Fallbacks = nil
	cfg.DialFunc = pgconn.DialFunc(dial)
	cfg.LookupFunc = func(_ context.Context, host string) ([]string, error) { return []string{host}, nil }
	cfg.ConnectTimeout = 3 * time.Second
	c, err := pgconn.ConnectConfig(ctx, cfg)
	if err != nil {
		return nil, err
	}
	defer c.Close(context.Background())
	r, err := c.Exec(ctx, "SELECT pg_is_in_recovery()").ReadAll()
	if err != nil || len(r) == 0 || len(r[0].Rows) == 0 {
		return nil, errors.New("PostgreSQL role unavailable")
	}
	info := singleInfo(p, "unknown", "primary")
	if string(r[0].Rows[0][0]) == "t" {
		info.Type = "pgsql-replication"
		info.Selected.Role = "replica"
		info.Nodes[0].Role = "replica"
		rows, e := c.Exec(ctx, "SELECT sender_host, sender_port FROM pg_stat_wal_receiver").ReadAll()
		if e == nil && len(rows) > 0 {
			for _, row := range rows[0].Rows {
				if len(row) < 2 {
					continue
				}
				port, _ := strconv.Atoi(string(row[1]))
				info.Nodes = addNodes(info.Nodes, model.ClusterNode{Host: string(row[0]), Port: port, Role: "primary"})
			}
		}
	} else {
		rows, e := c.Exec(ctx, "SELECT count(*) FROM pg_stat_replication").ReadAll()
		if e == nil && len(rows) > 0 && len(rows[0].Rows) > 0 && string(rows[0].Rows[0][0]) != "0" {
			info.Type = "pgsql-replication"
		}
	}
	if info.Type == "unknown" {
		info.Warning = "当前节点可写；没有可见复制连接时，无法区分单机与暂无副本的主节点。"
	}
	return info, nil
}

type sqlDialer struct {
	contextDialer
	host string
}

func (d sqlDialer) HostName() string { return d.host }
func probeSQLServer(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	tls, err := transport.TLSConfig(p)
	if err != nil {
		return nil, err
	}
	encryption := msdsn.Encryption(msdsn.EncryptionDisabled)
	if p.TargetTLS {
		encryption = msdsn.EncryptionRequired
	}
	c := mssql.NewConnectorConfig(msdsn.Config{Host: p.TargetHost, Port: uint64(p.TargetPort), User: p.TargetUser, Password: p.TargetPassword, Database: p.AuthSource, Encryption: encryption, TLSConfig: tls, Protocols: []string{"tcp"}, DisableRetry: true})
	c.Dialer = sqlDialer{contextDialer{dial}, p.TargetHost}
	db := sql.OpenDB(c)
	defer db.Close()
	var hadr, fc int
	var primary sql.NullInt64
	err = db.QueryRowContext(ctx, "SELECT CAST(SERVERPROPERTY('IsHadrEnabled') AS int), CAST(SERVERPROPERTY('IsClustered') AS int), sys.fn_hadr_is_primary_replica(DB_NAME())").Scan(&hadr, &fc, &primary)
	if err != nil {
		if err = db.PingContext(ctx); err != nil {
			return nil, err
		}
		return singleInfo(p, "unknown", "unknown"), nil
	}
	info := singleInfo(p, "standalone", "standalone")
	if fc == 1 {
		info.Type = "sqlserver-fci"
		info.Selected.Role = "primary"
	}
	if primary.Valid {
		info.Type = "sqlserver-ag"
		info.Selected.Role = "replica"
		if primary.Int64 == 1 {
			info.Selected.Role = "primary"
		}
		rows, e := db.QueryContext(ctx, "SELECT dns_name, port FROM sys.availability_group_listeners WHERE group_id IN (SELECT group_id FROM sys.availability_databases_cluster WHERE database_name = DB_NAME())")
		if e == nil {
			defer rows.Close()
			for rows.Next() {
				var host string
				var port int
				if rows.Scan(&host, &port) == nil {
					info.Nodes = addNodes(info.Nodes, model.ClusterNode{Host: host, Port: port, Role: "primary"})
				}
			}
		}
	}
	if hadr == 1 && !primary.Valid && fc == 0 {
		info.Type = "unknown"
		info.Warning = "已启用 Always On，但当前数据库不在可识别的可用性组中。"
	}
	info.Nodes[0].Role = info.Selected.Role
	return info, nil
}

func probeRabbit(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	result, err := amqpprobe.Probe(ctx, p, dial)
	if err != nil {
		return nil, err
	}
	info := singleInfo(p, "unknown", "broker")
	info.Name, info.Version = result.ClusterName, result.Version
	info.Warning = "AMQP 不公布集群成员列表；请配置其他集群节点以支持新连接故障切换。"
	if p.Cluster.Mode == "rabbitmq-cluster" {
		info.Type = "rabbitmq-cluster"
	}
	return info, nil
}