package db

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"strings"
	"time"

	puresqlite "github.com/glebarez/sqlite"
	"github.com/pingcap/agent-git-service/dbdsn"
	"gorm.io/driver/mysql"
	"gorm.io/driver/postgres"
	"gorm.io/gorm"
	gormlogger "gorm.io/gorm/logger"
)

// gormSlogLogger implements gormlogger.Interface using slog.
// This is inlined here to avoid db depending on internal/logging.
type gormSlogLogger struct {
	cfg gormlogger.Config
}

func (l *gormSlogLogger) Config() gormlogger.Config {
	return l.cfg
}

func (l *gormSlogLogger) LogMode(level gormlogger.LogLevel) gormlogger.Interface {
	clone := *l
	clone.cfg.LogLevel = level
	return &clone
}

func (l *gormSlogLogger) Info(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel < gormlogger.Info {
		return
	}
	slog.InfoContext(ctx, "gorm info",
		"component", "gorm",
		"message", fmt.Sprintf(msg, data...),
	)
}

func (l *gormSlogLogger) Warn(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel < gormlogger.Warn {
		return
	}
	slog.WarnContext(ctx, "gorm warning",
		"component", "gorm",
		"message", fmt.Sprintf(msg, data...),
	)
}

func (l *gormSlogLogger) Error(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel < gormlogger.Error {
		return
	}
	slog.ErrorContext(ctx, "gorm error",
		"component", "gorm",
		"message", fmt.Sprintf(msg, data...),
	)
}

func (l *gormSlogLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
	if l.cfg.LogLevel == gormlogger.Silent {
		return
	}
	if errors.Is(err, context.Canceled) {
		return
	}

	elapsed := time.Since(begin)
	switch {
	case err != nil && l.cfg.LogLevel >= gormlogger.Error && (!errors.Is(err, gorm.ErrRecordNotFound) || !l.cfg.IgnoreRecordNotFoundError):
		sql, rows := fc()
		slog.ErrorContext(ctx, "gorm query failed",
			"component", "gorm",
			"elapsed_ms", elapsed.Milliseconds(),
			"rows_affected", normalizeRows(rows),
			"sql", sanitizeSQL(sql),
			"error", err,
		)
	case l.cfg.SlowThreshold > 0 && elapsed > l.cfg.SlowThreshold && l.cfg.LogLevel >= gormlogger.Warn:
		sql, rows := fc()
		slog.WarnContext(ctx, "gorm slow query",
			"component", "gorm",
			"elapsed_ms", elapsed.Milliseconds(),
			"slow_threshold_ms", l.cfg.SlowThreshold.Milliseconds(),
			"rows_affected", normalizeRows(rows),
			"sql", sanitizeSQL(sql),
		)
	case l.cfg.LogLevel >= gormlogger.Info:
		sql, rows := fc()
		slog.InfoContext(ctx, "gorm query",
			"component", "gorm",
			"elapsed_ms", elapsed.Milliseconds(),
			"rows_affected", normalizeRows(rows),
			"sql", sanitizeSQL(sql),
		)
	}
}

func normalizeRows(rows int64) any {
	if rows < 0 {
		return nil
	}
	return rows
}

func sanitizeSQL(sql string) string {
	return strings.Join(strings.Fields(strings.TrimSpace(sql)), " ")
}

// Init opens the database connection and applies migrations.
func Init(dsn string) (*gorm.DB, error) {
	return open(dsn, true)
}

// Open opens the database connection without applying migrations.
func Open(dsn string) (*gorm.DB, error) {
	return open(dsn, false)
}

func open(dsn string, migrate bool) (*gorm.DB, error) {
	// Wrap the entire DB init in a timeout context to prevent indefinite hangs
	// when TiDB Cloud is slow to authenticate (e.g., cluster waking up).
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	var database *gorm.DB

	type result struct {
		db      *gorm.DB
		dialect string
		err     error
	}

	ch := make(chan result, 1)
	go func() {
		dialector, dialect := dialectorForDSN(dsn)
		cfg := &gorm.Config{
			Logger: &gormSlogLogger{cfg: gormlogger.Config{
				LogLevel:                  gormlogger.Warn,
				Colorful:                  false,
				ParameterizedQueries:      true,
				IgnoreRecordNotFoundError: true,
				SlowThreshold:             200 * time.Millisecond,
			}},
		}
		// PostgreSQL: PrepareStmt prevents the migrator from forcing
		// pgx.QueryExecModeSimpleProtocol, which can reject LIMIT params.
		if dialect == "postgres" {
			cfg.PrepareStmt = true
		}
		db, err := gorm.Open(dialector, cfg)
		if err != nil {
			err = fmt.Errorf("%s: %w", dialect, err)
		}
		ch <- result{db: db, dialect: dialect, err: err}
	}()

	var dialect string
	select {
	case <-ctx.Done():
		// Drain late result so the goroutine isn't blocked forever on ch<-,
		// and close any connection that completes after the timeout.
		go func() {
			if r := <-ch; r.db != nil {
				if sqlDB, err := r.db.DB(); err == nil {
					sqlDB.Close()
				}
			}
		}()
		return nil, fmt.Errorf("db: connection timed out after 30s (check DB_DSN and TiDB cluster status)")
	case r := <-ch:
		if r.err != nil {
			return nil, fmt.Errorf("db: failed to open: %w", r.err)
		}
		database = r.db
		dialect = r.dialect
	}

	var sqlDB *sql.DB
	if sqlDB, err := database.DB(); err == nil {
		configureConnectionPool(sqlDB, dialect, false)
	} else {
		slog.Warn("db: failed to get underlying sql.DB for pool configuration", "error", err)
	}

	if migrate {
		// TiDB Cloud serverless can take several minutes to wake and migrate.
		migrateCtx, migrateCancel := context.WithTimeout(context.Background(), 600*time.Second)
		defer migrateCancel()

		migrateDone := make(chan error, 1)
		go func() {
			migrateDone <- Migrate(database)
		}()

		select {
		case <-migrateCtx.Done():
			return nil, fmt.Errorf("db: auto-migrate timed out after 10m")
		case err := <-migrateDone:
			if err != nil {
				return nil, fmt.Errorf("db: auto-migrate failed: %w", err)
			}
		}
	}
	if sqlDB != nil {
		configureConnectionPool(sqlDB, dialect, true)
	}

	if migrate {
		slog.Info("db: connected and migrated.")
	} else {
		slog.Info("db: connected without migration.")
	}
	return database, nil
}

func configureConnectionPool(sqlDB *sql.DB, dialect string, migrated bool) {
	if sqlDB == nil {
		return
	}
	if dialect == "sqlite" {
		sqlDB.SetMaxOpenConns(1)
		sqlDB.SetMaxIdleConns(1)
		if !migrated {
			slog.Info("db: sqlite connection pool configured", "maxIdle", 1, "maxOpen", 1)
		}
		return
	}

	if migrated {
		return
	}
	// Configure connection pool tuning for TiDB/MySQL/PostgreSQL to prevent
	// PD/server timeouts without serializing normal multi-connection databases.
	sqlDB.SetMaxIdleConns(10)
	sqlDB.SetMaxOpenConns(50)
	sqlDB.SetConnMaxIdleTime(5 * time.Minute)
	sqlDB.SetConnMaxLifetime(30 * time.Minute)
	slog.Info("db: connection pool configured", "maxIdle", 10, "maxOpen", 50, "maxIdleTime", "5m", "maxLifetime", "30m")
}

func dialectorForDSN(dsn string) (gorm.Dialector, string) {
	raw := strings.TrimSpace(dsn)
	lower := strings.ToLower(raw)
	switch {
	case lower == ":memory:":
		return puresqlite.Open(normalizeSQLiteDSN(raw)), "sqlite"
	case strings.HasPrefix(lower, "file:"):
		return puresqlite.Open(normalizeSQLiteDSN(raw)), "sqlite"
	case strings.HasPrefix(lower, "sqlite://"):
		return puresqlite.Open(normalizeSQLiteDSN(raw[len("sqlite://"):])), "sqlite"
	case strings.HasPrefix(lower, "sqlite:"):
		trimmed := raw[len("sqlite:"):]
		trimmed = strings.TrimPrefix(trimmed, "//")
		return puresqlite.Open(normalizeSQLiteDSN(trimmed)), "sqlite"
	case strings.HasPrefix(lower, "postgres://") || strings.HasPrefix(lower, "postgresql://"):
		return postgres.Open(raw), "postgres"
	default:
		return mysql.Open(dbdsn.DefaultMySQLParseTime(raw)), "mysql"
	}
}

func normalizeSQLiteDSN(dsn string) string {
	parsed, err := url.Parse(dsn)
	if err != nil {
		return dsn
	}
	query := parsed.Query()
	pragmas := append([]string(nil), query["_pragma"]...)
	for _, compatibility := range []struct {
		key    string
		pragma string
	}{
		{key: "_busy_timeout", pragma: "busy_timeout"},
		{key: "_foreign_keys", pragma: "foreign_keys"},
		{key: "_journal_mode", pragma: "journal_mode"},
		{key: "_synchronous", pragma: "synchronous"},
	} {
		for _, value := range query[compatibility.key] {
			pragmas = append(pragmas, compatibility.pragma+"("+value+")")
		}
		query.Del(compatibility.key)
	}
	if len(pragmas) > 0 {
		query["_pragma"] = pragmas
	}
	parsed.RawQuery = query.Encode()
	return parsed.String()
}

// DialectorForDSN returns the GORM dialector and dialect name for the given DSN.
// Dialect name is one of "postgres", "mysql", "sqlite".
// Used by callers outside this package that need dialect-aware DB opens.
func DialectorForDSN(raw string) (gorm.Dialector, string) {
	return dialectorForDSN(raw)
}

// Migrate runs AutoMigrate for all application models on the given DB.
// It is called by Init and can be reused by tests or embedders.
func Migrate(database *gorm.DB) error {
	if err := MigrateWikiSlugColumnsBeforeAutoMigrate(database); err != nil {
		return err
	}
	if err := database.AutoMigrate(
		&User{},
		&OrganizationMember{},
		&OrganizationInvitation{},
		&OutsideCollaborator{},
		&BusinessUserActivityDaily{},
		&BusinessRepoActivityDaily{},
		&BusinessFirstUserAction{},
		&BusinessFirstRepoAction{},
		&AgentBinding{},
		&AgentInvite{},
		&Team{},
		&TeamMember{},
		&TeamRepository{},
		&Repository{},
		&RepositoryMigrationFreeze{},
		&RepositoryPushRecovery{},
		&RepositoryPushOutbox{},
		&RepositoryWebhookFanout{},
		&IssuePRNumberCounter{},
		&RepoRedirect{},
		&Token{},
		&UserIdentity{},
		&DeviceCode{},
		&DeviceCodeAuditLog{},
		&AuthorizationCode{},
		&Label{},
		&DeployKey{},
		&Milestone{},
		&Issue{},
		&IssueComment{},
		&IssueEvent{},
		&IssueReference{},
		&PagesConfig{},
		&PagesBuild{},
		&AuditLogEntry{},
		&PullRequest{},
		&LinkedBranch{},
		&Release{},
		&ReleaseAsset{},
		&Variable{},
		&Secret{},
		&Environment{},
		&SSHKey{},
		&GPGKey{},
		&SSHSigningKey{},
		&Project{},
		&ProjectField{},
		&ProjectItem{},
		&ProjectRepoLink{},
		&Workflow{},
		&WorkflowRun{},
		&WorkflowRunJob{},
		&Artifact{},
		&ActionCache{},
		&Ruleset{},
		&ReviewRequest{},
		&PullRequestReview{},
		&PRReviewComment{},
		&Autolink{},
		&Star{},
		&Gist{},
		&CommitStatus{},
		&Webhook{},
		&HookDelivery{},
		&Deployment{},
		&DeploymentStatus{},
		&BranchProtection{},
		&DependabotAlert{},
		&RepositoryInvitation{},
		&Collaborator{},
		&Reaction{},
		&Notification{},
		&WikiPageLabel{},
		&WikiPageIndex{},
		&WikiIndexState{},
		&WikiBacklink{},
		&WikiPageHistory{},
		&WikiSearchDocument{},
		&WikiPage{},
		&WikiPageRevision{},
		&WikiChangeset{},
		&WikiRepoHead{},
		&WikiCompactionJob{},
		&WikiDirIndex{},
		&WikiPageLink{},
		&WikiBlobRef{},
		&WikiPendingBlob{},
	); err != nil {
		return err
	}
	if err := MigrateIssueCommentPinningColumns(database); err != nil {
		return err
	}
	if err := MigrateIssueCommentThreadingColumns(database); err != nil {
		return err
	}
	// Enforce the user_kind column contract.
	if err := MigrateUserKind(database); err != nil {
		return err
	}
	if err := MigrateTeamOrgSlugIndex(database); err != nil {
		return err
	}
	if err := MigrateTeamMemberLookupIndexes(database); err != nil {
		return err
	}
	if err := MigrateLabelLookupIndexes(database); err != nil {
		return err
	}
	if err := MigrateIssueSearch(database); err != nil {
		return err
	}
	if err := MigrateWikiSearch(database); err != nil {
		return err
	}
	if err := MigrateWikiSlugColumns(database); err != nil {
		return err
	}
	if err := MigrateWikiChangesetSourceGit(database); err != nil {
		return err
	}
	// Add unique index on (project_id, content_id, type) to prevent duplicate items
	return MigrateProjectItemUniqueIndex(database)
}

// InitVector attempts to add VECTOR(dims) columns to the issues and
// pull_requests tables for semantic search, then best-effort vector indexes
// for TiDB-backed deployments that expose VEC_COSINE_DISTANCE. It uses raw SQL
// because GORM does not natively support TiDB's VECTOR type.
//
// This is a best-effort operation: unsupported deployments are skipped by
// capability probe, and DDL errors are logged without blocking startup.
func InitVector(database *gorm.DB, dims int) {
	if dims <= 0 || !SupportsVectorDistance(database) {
		return
	}
	// Cap at 16384 — TiDB supports up to ~16K dimensions.
	// This prevents a misconfigured Embedder.Dimensions() from creating
	// an absurd VECTOR(9223372036854775807) column that TiDB would reject.
	if dims > 16384 {
		slog.Warn("db: InitVector: dims capped to 16384", "requested", dims)
		dims = 16384
	}
	migrator := database.Migrator()
	tables := []string{"issues", "pull_requests"}
	for _, table := range tables {
		if migrator.HasColumn(table, "embedding") {
			slog.Debug("db: InitVector: embedding column already exists", "table", table)
			continue
		}
		sql := fmt.Sprintf("ALTER TABLE `%s` ADD COLUMN `embedding` VECTOR(%d)", table, dims)
		if err := database.Exec(sql).Error; err != nil {
			msg := strings.ToLower(err.Error())
			if migrator.HasColumn(table, "embedding") ||
				strings.Contains(msg, "duplicate column") ||
				strings.Contains(msg, "already exists") {
				slog.Debug("db: InitVector: embedding column already exists", "table", table)
				continue
			}
			slog.Warn("db: InitVector: "+table, "error", err)
		} else {
			slog.Debug("db: InitVector: added embedding column", "table", table, "dims", dims)
		}
	}

	ensureWikiSearchVector(database, dims)
	ensureVectorIndexes(database)
}