// Package database owns MeshX-local database initialization.
package database

import (
	"context"
	"crypto/sha1"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"time"

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

	"github.com/tidbcloud/AgentTeam/internal/store/content"
	"github.com/tidbcloud/AgentTeam/internal/store/content/migration"
	"github.com/tidbcloud/AgentTeam/internal/store/kb"
	"github.com/tidbcloud/AgentTeam/internal/store/knowledgeqa"
	"github.com/tidbcloud/AgentTeam/internal/store/providertokens"
	"github.com/tidbcloud/AgentTeam/internal/store/repocache"
	"github.com/tidbcloud/AgentTeam/internal/store/repotokens"
	"github.com/tidbcloud/AgentTeam/internal/store/resource/activity"
	"github.com/tidbcloud/AgentTeam/internal/store/resource/memory"
	"github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
	rolesstore "github.com/tidbcloud/AgentTeam/internal/store/resource/roles"
	skillsstore "github.com/tidbcloud/AgentTeam/internal/store/resource/skills"
)

const (
	meshxMigrationLockWait       = 15 * time.Second
	meshxMigrationReleaseTimeout = 5 * time.Second
)

var meshxMigrationProcessLock = make(chan struct{}, 1)

// Config describes the MeshX-owned database. It is intentionally separate
// from the embedded AGS database configuration.
type Config struct {
	DSN string
}

type slogLogger struct{ cfg gormlogger.Config }

func (l *slogLogger) LogMode(level gormlogger.LogLevel) gormlogger.Interface {
	clone := *l
	clone.cfg.LogLevel = level
	return &clone
}
func (l *slogLogger) Info(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel >= gormlogger.Info {
		slog.InfoContext(ctx, "meshx gorm info", "message", fmt.Sprintf(msg, data...))
	}
}
func (l *slogLogger) Warn(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel >= gormlogger.Warn {
		slog.WarnContext(ctx, "meshx gorm warning", "message", fmt.Sprintf(msg, data...))
	}
}
func (l *slogLogger) Error(ctx context.Context, msg string, data ...interface{}) {
	if l.cfg.LogLevel >= gormlogger.Error {
		slog.ErrorContext(ctx, "meshx gorm error", "message", fmt.Sprintf(msg, data...))
	}
}
func (l *slogLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
	if l.cfg.LogLevel == gormlogger.Silent || errors.Is(err, context.Canceled) {
		return
	}
	elapsed := time.Since(begin)
	if err != nil && l.cfg.LogLevel >= gormlogger.Error && (!errors.Is(err, gorm.ErrRecordNotFound) || !l.cfg.IgnoreRecordNotFoundError) {
		sql, rows := fc()
		slog.ErrorContext(ctx, "meshx gorm query failed", "elapsed_ms", elapsed.Milliseconds(), "rows_affected", normalizeRows(rows), "sql", sanitizeSQL(sql), "error", err)
		return
	}
	if l.cfg.SlowThreshold > 0 && elapsed > l.cfg.SlowThreshold && l.cfg.LogLevel >= gormlogger.Warn {
		sql, rows := fc()
		slog.WarnContext(ctx, "meshx gorm slow query", "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)), " ") }

func open(cfg Config, migrate bool) (*gorm.DB, error) {
	dsn := strings.TrimSpace(cfg.DSN)
	if dsn == "" {
		return nil, fmt.Errorf("meshx database DSN is required")
	}
	if err := ensureSQLiteDir(dsn); err != nil {
		return nil, err
	}
	dialector, dialect := dialectorForDSN(dsn)
	gcfg := &gorm.Config{Logger: &slogLogger{cfg: gormlogger.Config{LogLevel: gormlogger.Warn, Colorful: false, ParameterizedQueries: true, IgnoreRecordNotFoundError: true, SlowThreshold: 200 * time.Millisecond}}}
	if dialect == "postgres" {
		gcfg.PrepareStmt = true
	}
	db, err := gorm.Open(dialector, gcfg)
	if err != nil {
		return nil, fmt.Errorf("open meshx database (%s): %w", dialect, err)
	}
	if sqlDB, err := db.DB(); err == nil {
		sqlDB.SetMaxIdleConns(5)
		sqlDB.SetMaxOpenConns(20)
		sqlDB.SetConnMaxIdleTime(5 * time.Minute)
		sqlDB.SetConnMaxLifetime(30 * time.Minute)
	}
	if migrate {
		if err := Migrate(db); err != nil {
			_ = Close(db)
			return nil, err
		}
	}
	return db, nil
}

// Open opens and migrates the MeshX-local database.
func Open(cfg Config) (*gorm.DB, error) { return open(cfg, true) }

// OpenWithoutMigrate opens an existing database without applying MeshX migrations.
func OpenWithoutMigrate(cfg Config) (*gorm.DB, error) { return open(cfg, false) }

// Close closes the underlying SQL database.
func Close(db *gorm.DB) error {
	if db == nil {
		return nil
	}
	sqlDB, err := db.DB()
	if err != nil {
		return err
	}
	return sqlDB.Close()
}

// Migrate runs MeshX-local schema migrations.
func Migrate(db *gorm.DB) error {
	if db == nil {
		return fmt.Errorf("meshx database is nil")
	}
	return withMeshXMigrationLock(context.Background(), db, migrateLocked)
}

func migrateLocked(db *gorm.DB) error {
	if err := prepareKnowledgeBaseSetupSessionSnapshotColumns(db); err != nil {
		return fmt.Errorf("prepare knowledge base setup session snapshot columns: %w", err)
	}
	if err := migrateContentAuthoritySchema(db); err != nil {
		return fmt.Errorf("migrate content authority schema: %w", err)
	}
	if err := db.AutoMigrate(
		&content.VersionedContent{},
		&content.ContentVersion{},
		&content.VersionOperation{},
		&content.VersionPath{},
		&content.VersionPin{},
		&content.Outbox{},
		&content.StagedPayloadGC{},
		&migration.Mapping{},
		&migration.Job{},
		&kb.KnowledgeBase{},
		&kb.KnowledgeBaseGroup{},
		&kb.KnowledgeBaseTag{},
		&kb.KnowledgeBaseTagAssignment{},
		&kb.KnowledgeBasePin{},
		&kb.KnowledgeBaseGrant{},
		&kb.KnowledgeBaseSourceRepositoryAccessOverride{},
		&kb.KnowledgeBaseSetupSession{},
		&kb.KnowledgeBaseSetupSourceArtifact{},
		&kb.KnowledgeBaseGenerationRequest{},
		&kb.KnowledgeBasePublicationRequest{},
		&kb.Source{},
		&kb.SourceMirror{},
		&kb.AnalysisRun{},
		&kb.SourceRevision{},
		&kb.SourceContentMapping{},
		&kb.SourceSearchDocument{},
		&kb.Job{},
		&kb.AuditEvent{},
		&repocache.Cache{},
		&kb.KBSavedView{},
		&knowledgeqa.RootHandleRecord{},
		&knowledgeqa.AccountingCheckpointRecord{},
		&providertokens.PersonalAccessToken{},
		&repotokens.Token{},
		&repotokens.Event{},
		&repotokens.AgentSuppression{},
		&activity.ResourceActivityEvent{},
		&activity.ResourceActivityLatestCommit{},
		&activity.ResourceActivityProjectorCheckpoint{},
		&memory.Memory{},
		&memory.MemoryContentCheckpoint{},
		&memory.MemoryGrant{},
		&memory.MemoryPin{},
		&memory.AgentAssetAccessChangeOutbox{},
		&skillsstore.Skill{},
		&skillsstore.SkillVersion{},
		&skillsstore.SkillRepositoryBranch{},
		&skillsstore.SkillAgentBinding{},
		&skillsstore.SkillGrant{},
		&rolesstore.Role{},
		&rolesstore.RoleVersion{},
		&rolesstore.RolePackageManifestItem{},
		&rolesstore.RoleGrant{},
		&rolesstore.RoleRealization{},
		&rolesstore.RoleRealizationItem{},
		&rolesstore.AgentRoleBinding{},
		&rolesstore.RoleOperationReceipt{},
		&repos.ResourceRepository{},
		&repos.ResourceRepositoryLifecycleEvent{},
		&repos.ResourceRepositoryAccessGrant{},
	); err != nil {
		return fmt.Errorf("migrate meshx database: %w", err)
	}
	if err := backfillKnowledgeBaseGroups(db); err != nil {
		return fmt.Errorf("backfill knowledge base groups: %w", err)
	}
	if err := backfillResourceActivityLatestCommits(db); err != nil {
		return fmt.Errorf("backfill latest resource activity commits: %w", err)
	}
	if err := dropSkillAgentBindingRepoSnapshotColumns(db); err != nil {
		return fmt.Errorf("drop meshx skill agent binding repo snapshot columns: %w", err)
	}
	if err := migrateSkillVisibilityScope(db); err != nil {
		return fmt.Errorf("migrate meshx skill visibility scope: %w", err)
	}
	if err := backfillSkillOwnerGrants(db); err != nil {
		return fmt.Errorf("backfill meshx skill owner grants: %w", err)
	}
	if err := backfillRoleOwnerGrants(db); err != nil {
		return fmt.Errorf("backfill meshx role owner grants: %w", err)
	}
	if err := applyAutomationMigrations(db); err != nil {
		return fmt.Errorf("migrate meshx automation schema: %w", err)
	}
	if err := VerifySchema(db); err != nil {
		return fmt.Errorf("verify meshx schema: %w", err)
	}
	return nil
}

// migrateContentAuthoritySchema avoids sending the retired authority schema
// through SQLite's column-by-column table rebuild. That rebuild omits the
// activation checkpoint from its first copy. Migration 0025 consumes and
// removes the legacy marker columns before later startups return to normal
// AutoMigrate; fresh and already-current schemas use that path immediately.
func migrateContentAuthoritySchema(db *gorm.DB) error {
	if hasLegacyContentAuthorityColumns(db) {
		if !db.Migrator().HasIndex(&content.Authority{}, "uk_content_authority_binding") {
			return db.Migrator().CreateIndex(&content.Authority{}, "uk_content_authority_binding")
		}
		return nil
	}
	return db.AutoMigrate(&content.Authority{})
}

func prepareKnowledgeBaseSetupSessionSnapshotColumns(db *gorm.DB) error {
	if !db.Migrator().HasTable(&kb.KnowledgeBaseSetupSession{}) {
		return nil
	}
	for _, column := range []string{"working_snapshot_json", "published_snapshot_json"} {
		if db.Migrator().HasColumn(&kb.KnowledgeBaseSetupSession{}, column) {
			continue
		}
		if db.Dialector.Name() == "sqlite" {
			if err := db.Exec("ALTER TABLE knowledge_base_setup_sessions ADD COLUMN " + column + " text NOT NULL DEFAULT ''").Error; err != nil {
				return err
			}
			continue
		}
		if err := db.Exec("ALTER TABLE knowledge_base_setup_sessions ADD COLUMN " + column + " text NULL").Error; err != nil {
			return err
		}
		if err := db.Exec("UPDATE knowledge_base_setup_sessions SET " + column + " = '' WHERE " + column + " IS NULL").Error; err != nil {
			return err
		}
		if err := db.Exec("ALTER TABLE knowledge_base_setup_sessions MODIFY COLUMN " + column + " text NOT NULL").Error; err != nil {
			return err
		}
	}
	return nil
}

func backfillKnowledgeBaseGroups(db *gorm.DB) error {
	// Keep this repair idempotent rather than one-time: during a rolling deploy,
	// an older process can still insert a KB with the column default after a
	// newer process has migrated the schema.
	return db.Transaction(func(tx *gorm.DB) error {
		var workspaceIDs []string
		if err := tx.Model(&kb.KnowledgeBase{}).
			Distinct("workspace_id").
			Where("workspace_id <> ?", "").
			Pluck("workspace_id", &workspaceIDs).Error; err != nil {
			return err
		}
		now := time.Now().UTC()
		for _, workspaceID := range workspaceIDs {
			workspaceID = strings.TrimSpace(workspaceID)
			if workspaceID == "" {
				continue
			}
			groupID := kb.DefaultKnowledgeBaseGroupID(workspaceID)
			group := kb.KnowledgeBaseGroup{
				ID:          groupID,
				WorkspaceID: workspaceID,
				Name:        "Uncategorized",
				Description: "Knowledge bases that have not been assigned to another group.",
				Color:       "#737373",
				SortOrder:   10000,
				IsDefault:   true,
				CreatedBy:   "system",
				UpdatedBy:   "system",
				CreatedAt:   now,
				UpdatedAt:   now,
			}
			if err := tx.Where("id = ?", groupID).FirstOrCreate(&group).Error; err != nil {
				return err
			}
			if err := tx.Model(&kb.KnowledgeBase{}).
				Where("workspace_id = ? AND (group_id = ? OR group_id IS NULL)", workspaceID, "").
				Update("group_id", groupID).Error; err != nil {
				return err
			}
		}
		return nil
	})
}

func backfillResourceActivityLatestCommits(db *gorm.DB) error {
	return applyOneTimeMeshXMigration(db, "resource_activity_latest_commits_backfill", func(tx *gorm.DB) error {
		insert := "INSERT INTO"
		suffix := ""
		switch strings.ToLower(strings.TrimSpace(tx.Dialector.Name())) {
		case "mysql":
			insert = "INSERT IGNORE INTO"
		case "sqlite":
			insert = "INSERT OR IGNORE INTO"
		case "postgres":
			suffix = " ON CONFLICT DO NOTHING"
		}
		return tx.Exec(insert + ` resource_activity_latest_commits
		(workspace_id, asset_type, asset_id, occurred_at, event_id)
	SELECT workspace_id, asset_type, asset_id, occurred_at, id
	FROM (
		SELECT workspace_id, asset_type, asset_id, occurred_at, id,
			ROW_NUMBER() OVER (
				PARTITION BY workspace_id, asset_type, asset_id
				ORDER BY occurred_at DESC, id DESC
			) AS rn
		FROM resource_activity_events
		WHERE event_type = 'commit_pushed'
	) ranked_resource_activity_events
	WHERE rn = 1` + suffix).Error
	})
}

func applyOneTimeMeshXMigration(db *gorm.DB, name string, apply func(*gorm.DB) error) error {
	return db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Exec(`CREATE TABLE IF NOT EXISTS meshx_data_migrations (
			name VARCHAR(128) NOT NULL,
			applied_at TIMESTAMP NOT NULL,
			PRIMARY KEY (name)
		)`).Error; err != nil {
			return err
		}
		var applied int64
		if err := tx.Table("meshx_data_migrations").Where("name = ?", name).Count(&applied).Error; err != nil || applied > 0 {
			return err
		}
		if err := apply(tx); err != nil {
			return err
		}
		return tx.Exec("INSERT INTO meshx_data_migrations (name, applied_at) VALUES (?, ?)", name, time.Now().UTC()).Error
	})
}

func withMeshXMigrationLock(ctx context.Context, db *gorm.DB, fn func(*gorm.DB) error) (err error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if db.Dialector.Name() != "mysql" {
		select {
		case meshxMigrationProcessLock <- struct{}{}:
			defer func() { <-meshxMigrationProcessLock }()
		case <-ctx.Done():
			return ctx.Err()
		}
		return fn(db.WithContext(ctx))
	}
	sqlDB, err := db.DB()
	if err != nil {
		return fmt.Errorf("meshx migration sql db: %w", err)
	}
	conn, err := sqlDB.Conn(ctx)
	if err != nil {
		return fmt.Errorf("pin meshx migration session: %w", err)
	}
	defer func() { err = errors.Join(err, conn.Close()) }()
	locked := db.Session(&gorm.Session{NewDB: true, Context: ctx})
	locked.ConnPool = conn
	locked.Statement.ConnPool = conn
	lockName, err := meshxMigrationLockName(ctx, conn)
	if err != nil {
		return err
	}
	var acquired sql.NullInt64
	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", lockName, int(meshxMigrationLockWait/time.Second)).Scan(&acquired); err != nil {
		return fmt.Errorf("acquire meshx migration lock: %w", err)
	}
	if !acquired.Valid || acquired.Int64 != 1 {
		return fmt.Errorf("acquire meshx migration lock: timed out")
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), meshxMigrationReleaseTimeout)
		defer cancel()
		var released sql.NullInt64
		releaseErr := conn.QueryRowContext(releaseCtx, "SELECT RELEASE_LOCK(?)", lockName).Scan(&released)
		if releaseErr == nil && (!released.Valid || released.Int64 != 1) {
			releaseErr = fmt.Errorf("current session did not release meshx migration lock")
		}
		if releaseErr != nil {
			err = errors.Join(err, fmt.Errorf("release meshx migration lock: %w", releaseErr))
		}
	}()
	return fn(locked)
}

func meshxMigrationLockName(ctx context.Context, conn *sql.Conn) (string, error) {
	var databaseName sql.NullString
	if err := conn.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&databaseName); err != nil {
		return "", fmt.Errorf("detect meshx migration database: %w", err)
	}
	sum := sha1.Sum([]byte(strings.TrimSpace(databaseName.String)))
	return "meshx_migrate:" + hex.EncodeToString(sum[:8]), nil
}

func backfillSkillOwnerGrants(db *gorm.DB) error {
	var skills []skillsstore.Skill
	if err := db.Where("created_by <> ?", "").Find(&skills).Error; err != nil {
		return err
	}
	now := time.Now().UTC()
	for _, skill := range skills {
		var count int64
		if err := db.Model(&skillsstore.SkillGrant{}).
			Where("workspace_id = ? AND skill_id = ? AND subject_type = ? AND subject_id = ?", skill.WorkspaceID, skill.ID, "user", skill.CreatedBy).
			Count(&count).Error; err != nil {
			return err
		}
		if count > 0 {
			continue
		}
		grantTime := skill.CreatedAt
		if grantTime.IsZero() {
			grantTime = now
		}
		grant := skillsstore.SkillGrant{
			ID:          skillOwnerGrantBackfillID(skill),
			WorkspaceID: skill.WorkspaceID,
			SkillID:     skill.ID,
			SubjectType: "user",
			SubjectID:   skill.CreatedBy,
			Role:        "owner",
			CreatedBy:   skill.CreatedBy,
			UpdatedBy:   skill.CreatedBy,
			CreatedAt:   grantTime,
			UpdatedAt:   grantTime,
		}
		if err := db.Create(&grant).Error; err != nil {
			return err
		}
	}
	return nil
}

func skillOwnerGrantBackfillID(skill skillsstore.Skill) string {
	h := sha1.Sum([]byte(skill.WorkspaceID + "\x00" + skill.ID + "\x00" + skill.CreatedBy))
	return "skg_owner_" + hex.EncodeToString(h[:])[:32]
}

func backfillRoleOwnerGrants(db *gorm.DB) error {
	var roles []rolesstore.Role
	if err := db.Where("created_by <> ?", "").Find(&roles).Error; err != nil {
		return err
	}
	now := time.Now().UTC()
	for _, role := range roles {
		var count int64
		if err := db.Model(&rolesstore.RoleGrant{}).
			Where("workspace_id = ? AND role_id = ? AND subject_type = ? AND subject_id = ?", role.WorkspaceID, role.ID, "user", role.CreatedBy).
			Count(&count).Error; err != nil {
			return err
		}
		if count > 0 {
			continue
		}
		grantTime := role.CreatedAt
		if grantTime.IsZero() {
			grantTime = now
		}
		grant := rolesstore.RoleGrant{
			ID:          roleOwnerGrantBackfillID(role),
			WorkspaceID: role.WorkspaceID,
			RoleID:      role.ID,
			SubjectType: "user",
			SubjectID:   role.CreatedBy,
			Role:        "owner",
			CreatedBy:   role.CreatedBy,
			UpdatedBy:   role.CreatedBy,
			CreatedAt:   grantTime,
			UpdatedAt:   grantTime,
		}
		if err := db.Create(&grant).Error; err != nil {
			return err
		}
	}
	return nil
}

func roleOwnerGrantBackfillID(role rolesstore.Role) string {
	h := sha1.Sum([]byte(role.WorkspaceID + "\x00" + role.ID + "\x00" + role.CreatedBy))
	return "rgr_owner_" + hex.EncodeToString(h[:])[:32]
}

func dialectorForDSN(raw string) (gorm.Dialector, string) {
	lower := strings.ToLower(strings.TrimSpace(raw))
	switch {
	case lower == ":memory:":
		return puresqlite.Open(raw), "sqlite"
	case strings.HasPrefix(lower, "file:"):
		if isMemorySQLiteDSN(lower) {
			return puresqlite.Open(raw), "sqlite"
		}
		return cgosqlite.Open(raw), "sqlite"
	case strings.HasPrefix(lower, "sqlite://"):
		trimmed := raw[len("sqlite://"):]
		if isMemorySQLiteDSN(strings.ToLower(trimmed)) {
			return puresqlite.Open(trimmed), "sqlite"
		}
		return cgosqlite.Open(trimmed), "sqlite"
	case strings.HasPrefix(lower, "sqlite:"):
		trimmed := strings.TrimPrefix(raw[len("sqlite:"):], "//")
		if isMemorySQLiteDSN(strings.ToLower(trimmed)) {
			return puresqlite.Open(trimmed), "sqlite"
		}
		return cgosqlite.Open(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 isMemorySQLiteDSN(lower string) bool {
	lower = strings.TrimSpace(lower)
	return lower == ":memory:" || strings.Contains(lower, "mode=memory") || strings.HasPrefix(lower, "file::memory:")
}

func ensureSQLiteDir(dsn string) error {
	path := sqlitePath(dsn)
	if path == "" || path == ":memory:" || strings.HasPrefix(path, "file::memory:") {
		return nil
	}
	dir := filepath.Dir(path)
	if dir == "." || dir == "" {
		return nil
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return fmt.Errorf("create meshx database dir %q: %w", dir, err)
	}
	return nil
}

func sqlitePath(dsn string) string {
	raw := strings.TrimSpace(dsn)
	lower := strings.ToLower(raw)
	switch {
	case lower == ":memory:":
		return raw
	case strings.HasPrefix(lower, "file:"):
		p := strings.TrimPrefix(raw, "file:")
		if idx := strings.IndexAny(p, "?"); idx >= 0 {
			p = p[:idx]
		}
		return p
	case strings.HasPrefix(lower, "sqlite://"):
		p := raw[len("sqlite://"):]
		if idx := strings.IndexAny(p, "?"); idx >= 0 {
			p = p[:idx]
		}
		return p
	case strings.HasPrefix(lower, "sqlite:"):
		p := strings.TrimPrefix(raw[len("sqlite"):], ":")
		p = strings.TrimPrefix(p, "//")
		if idx := strings.IndexAny(p, "?"); idx >= 0 {
			p = p[:idx]
		}
		return p
	default:
		return ""
	}
}