logs/internal/models/schema_version.go

package models

import (
	"fmt"

	"gorm.io/gorm"
)

type schemaMigration struct {
	script   string
	checksum string
}

var requiredSchema = struct {
	migrations []schemaMigration
	tables     []string
}{
	migrations: []schemaMigration{
		{script: "0001_baseline.up.sql", checksum: "f6632f4927b205caf88bdbcf5f9db755ed4a174caacd4900d1aac12839fb9c5e"},
		{script: "0002_otlp_ingest_key.up.sql", checksum: "85923db07fea3e0dfaa33af0c2a3775b917c1463717c01300e0958602efeec1d"},
	},
	tables: []string{"logs_events", "logs_syslog_rules", "logs_trap_rules", "logs_trace_spans"},
}

// RequireSchemaVersion 确认当前数据库已经由受控迁移脚本初始化。
func RequireSchemaVersion(db *gorm.DB, service string) error {
	if db == nil {
		return fmt.Errorf("%s 数据库连接未初始化", service)
	}

	for _, migration := range requiredSchema.migrations {
		var count int64
		err := db.Raw(`
SELECT COUNT(*)
FROM public.ops_schema_migrations
WHERE service = ? AND script_name = ? AND checksum = ? AND status = 'success'`,
			service, migration.script, migration.checksum).Scan(&count).Error
		if err != nil {
			return fmt.Errorf("%s 数据库版本检查失败，请先执行 migrate-all.sh up --service %s: %w", service, service, err)
		}
		if count != 1 {
			return fmt.Errorf("%s 数据库迁移版本不匹配：%s，请执行 migrate-all.sh status --service %s", service, migration.script, service)
		}
	}

	var err error
	for _, table := range requiredSchema.tables {
		var valid bool
		err = db.Raw(`
SELECT to_regclass(?) IS NOT NULL AND EXISTS (
  SELECT 1
  FROM pg_constraint c
  JOIN pg_class t ON t.oid = c.conrelid
  JOIN pg_namespace n ON n.oid = t.relnamespace
  WHERE n.nspname = 'public' AND t.relname = ? AND c.contype = 'p'
)`, "public."+table, table).Scan(&valid).Error
		if err != nil || !valid {
			return fmt.Errorf("%s 数据库关键表结构不完整：%s", service, table)
		}
	}

	var ingestKeyValid bool
	err = db.Raw(`
SELECT EXISTS (
  SELECT 1
  FROM information_schema.columns
  WHERE table_schema = 'public'
    AND table_name = 'logs_events'
    AND column_name = 'ingest_key'
    AND data_type = 'character varying'
    AND character_maximum_length = 160
    AND is_nullable = 'YES'
) AND EXISTS (
  SELECT 1
  FROM pg_indexes
  WHERE schemaname = 'public'
    AND tablename = 'logs_events'
    AND indexname = 'idx_logs_events_ingest_key'
    AND indexdef LIKE 'CREATE UNIQUE INDEX %'
)`).Scan(&ingestKeyValid).Error
	if err != nil || !ingestKeyValid {
		return fmt.Errorf("%s 数据库关键字段或索引不完整：logs_events.ingest_key", service)
	}
	return nil
}