package tenants

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"

	_ "github.com/go-sql-driver/mysql"

	"github.com/tidbcloud/observability/vm-migration/internal/jsonio"
)

// defaultDBPort is the secret-file port fallback when the JSON omits one.
const defaultDBPort = 3306

// essentialDBBatch is the application_id IN-query batch size.
const essentialDBBatch = 500

// MySQLConfig is a resolved regional-server connection config.
type MySQLConfig struct {
	Host     string
	Port     int
	User     string
	Password string
	Name     string
	TLS      bool
}

// ResolveMySQLConfig resolves DB connection params from the --secret-file
// JSON. --db-host overrides the secret's host for port-forwarded access.
func ResolveMySQLConfig(o EssentialRdsOptions) (MySQLConfig, error) {
	if o.SecretFile == "" {
		return MySQLConfig{}, errors.New("--secret-file is required (RDS secret JSON path)")
	}
	var s struct {
		Host     string `json:"host"`
		Port     int    `json:"port"`
		Username string `json:"username"`
		Password string `json:"password"`
		Dbname   string `json:"dbname"`
	}
	if err := jsonio.ReadFile(o.SecretFile, &s); err != nil {
		return MySQLConfig{}, fmt.Errorf("read secret-file %s: %w", o.SecretFile, err)
	}
	if s.Port == 0 {
		s.Port = defaultDBPort
	}
	host := s.Host
	if o.DBHost != "" {
		host = o.DBHost
	}
	return MySQLConfig{Host: host, Port: s.Port, User: s.Username, Password: s.Password, Name: s.Dbname, TLS: o.TLS}, nil
}

// VMInfo is the joined clusters+cluster_specs row for one application_id.
type VMInfo struct {
	InternalClusterID string
	DeletionTime      sql.NullTime
	UpdateTime        sql.NullTime
	VMAccountID       sql.NullString
	VMProjectID       sql.NullString
}

// VMInfoQuerier abstracts the regional-server vm-info join so tests inject
// fakes without a MySQL connection.
type VMInfoQuerier interface {
	QueryVMInfo(ctx context.Context, appIDs []string) (map[string]VMInfo, error)
}

// mysqlQuerier is the production VMInfoQuerier.
type mysqlQuerier struct {
	db *sql.DB
}

// NewMySQLQuerier opens the regional-server MySQL connection and verifies it
// with a ping under ctx.
//
// Returns the production VMInfoQuerier plus a closer for the underlying
// connection; callers own both.
func NewMySQLQuerier(ctx context.Context, cfg MySQLConfig) (VMInfoQuerier, func() error, error) {
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=UTC",
		cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Name)
	if cfg.TLS {
		dsn += "&tls=skip-verify"
	}
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, nil, fmt.Errorf("open mysql: %w", err)
	}
	db.SetMaxOpenConns(4)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return nil, nil, fmt.Errorf("connect mysql: %w", err)
	}
	return &mysqlQuerier{db: db}, db.Close, nil
}

// QueryVMInfo batch-queries clusters+cluster_specs by application_id,
// keeping the best row per id (keepBestVMInfo).
func (m *mysqlQuerier) QueryVMInfo(ctx context.Context, appIDs []string) (map[string]VMInfo, error) {
	best := make(map[string]VMInfo, len(appIDs))
	for _, chunk := range batches(appIDs, essentialDBBatch) {
		// #nosec G202 -- placeholders are generated "?," repeats; all values
		// are passed as query args.
		q := `SELECT c.application_id,
		             c.cluster_id,
		             c.deletion_timestamp,
		             c.update_timestamp,
		             s.vm_account_id,
		             s.vm_project_id
		      FROM clusters c
		      LEFT JOIN cluster_specs s ON c.cluster_id = s.cluster_id
		      WHERE c.application_id IN (` + strings.TrimSuffix(strings.Repeat("?,", len(chunk)), ",") + `)`
		args := make([]any, len(chunk))
		for i, a := range chunk {
			args[i] = a
		}
		rows, err := m.db.QueryContext(ctx, q, args...)
		if err != nil {
			return nil, fmt.Errorf("query application_ids batch: %w", err)
		}
		for rows.Next() {
			var d VMInfo
			var appID string
			if err := rows.Scan(&appID, &d.InternalClusterID, &d.DeletionTime,
				&d.UpdateTime, &d.VMAccountID, &d.VMProjectID); err != nil {
				_ = rows.Close()
				return nil, fmt.Errorf("scan: %w", err)
			}
			keepBestVMInfo(best, appID, d)
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return nil, fmt.Errorf("rows iter: %w", err)
		}
		_ = rows.Close()
	}
	return best, nil
}

// keepBestVMInfo prefers rows with vm info populated; ties break by the
// latest update_timestamp.
func keepBestVMInfo(best map[string]VMInfo, appID string, d VMInfo) {
	cur, ok := best[appID]
	if !ok {
		best[appID] = d
		return
	}
	curHas := cur.VMAccountID.Valid && cur.VMAccountID.String != ""
	newHas := d.VMAccountID.Valid && d.VMAccountID.String != ""
	switch {
	case newHas && !curHas:
		best[appID] = d
	case newHas == curHas && vmInfoUpdatedAfter(d, cur):
		best[appID] = d
	}
}

func vmInfoUpdatedAfter(a, b VMInfo) bool {
	if !a.UpdateTime.Valid {
		return false
	}
	if !b.UpdateTime.Valid {
		return true
	}
	return a.UpdateTime.Time.After(b.UpdateTime.Time)
}

// batches splits s into consecutive chunks of at most size.
func batches(s []string, size int) [][]string {
	var out [][]string
	for i := 0; i < len(s); i += size {
		j := i + size
		if j > len(s) {
			j = len(s)
		}
		out = append(out, s[i:j])
	}
	return out
}