feat: add read-only schema preflight

internal/storage/mysql/operational_metrics_integration_test.go

@@ -24,6 +24,10 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
 	if err := Migrate(ctx, db); err != nil {
 		t.Fatal(err)
 	}
+	verification, err := VerifyOperationalSchema(ctx, db)
+	if err != nil || !verification.Complete || verification.ForeignKeyCount != 0 || verification.RequiredIndexCount != 6 {
+		t.Fatalf("schema preflight=%+v err=%v", verification, err)
+	}
 	if err := CheckSchema(ctx, db); err != nil {
 		t.Fatal(err)
 	}

internal/storage/mysql/schema_verify.go

@@ -0,0 +1,83 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+)
+
+// SchemaVerification 是发布和恢复前的只读结构核验结果。
+// 它不修改数据库，也不代表已经完成备份恢复或高可用验收。
+type SchemaVerification struct {
+	// SchemaVersion 是迁移账本中的最后版本。
+	SchemaVersion uint64 `json:"schema_version"`
+	// MigrationCount 是已记录且通过校验的迁移数量。
+	MigrationCount int `json:"migration_count"`
+	// ForeignKeyCount 是 Orbit 表上的数据库外键数量，必须为零。
+	ForeignKeyCount int `json:"foreign_key_count"`
+	// RequiredIndexCount 是有界观测所需固定索引数量，当前必须为六。
+	RequiredIndexCount int `json:"required_index_count"`
+	// Complete 表示迁移、外键和观测索引均满足当前二进制要求。
+	Complete bool `json:"complete"`
+}
+
+var requiredOperationalIndexes = []struct {
+	table string
+	name  string
+}{
+	{"pulse_tasks", "ix_task_observe_state_available"},
+	{"orbit_deliveries", "ix_delivery_observe_state_available"},
+	{"weave_instances", "ix_workflow_observe_state_updated"},
+	{"orbit_events", "ix_event_observe_type_occurred"},
+	{"pulse_tasks", "ix_task_observe_terminal_completed"},
+	{"weave_instances", "ix_workflow_observe_terminal_completed"},
+}
+
+// VerifyOperationalSchema performs the read-only preflight used before release
+// or restore. It never runs migration or repairs a dirty database.
+func VerifyOperationalSchema(ctx context.Context, db *sql.DB) (SchemaVerification, error) {
+	if db == nil {
+		return SchemaVerification{}, errors.New("database is required")
+	}
+	status, err := Status(ctx, db)
+	if err != nil {
+		return SchemaVerification{}, errors.New("schema status unavailable")
+	}
+	result := SchemaVerification{MigrationCount: len(status)}
+	if len(status) > 0 {
+		result.SchemaVersion = status[len(status)-1].Version
+	}
+	for _, item := range status {
+		if item.Dirty {
+			return result, errors.New("schema has a dirty migration")
+		}
+	}
+	var foreignKeys int
+	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE() AND LEFT(TABLE_NAME,6) IN ('orbit_','pulse_','weave_')`).Scan(&foreignKeys); err != nil {
+		return result, errors.New("foreign key status unavailable")
+	}
+	result.ForeignKeyCount = foreignKeys
+	for _, required := range requiredOperationalIndexes {
+		var count int
+		if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND INDEX_NAME=?`, required.table, required.name).Scan(&count); err != nil {
+			return result, errors.New("operational index status unavailable")
+		}
+		if count > 0 {
+			result.RequiredIndexCount++
+		}
+	}
+	wanted, err := Migrations()
+	if err != nil || len(status) != len(wanted) {
+		return result, errors.New("schema version mismatch")
+	}
+	for i, migration := range wanted {
+		if status[i].Version != migration.Version || status[i].Name != migration.Name || status[i].Checksum != migration.Checksum {
+			return result, errors.New("schema checksum mismatch")
+		}
+	}
+	result.Complete = result.ForeignKeyCount == 0 && result.RequiredIndexCount == len(requiredOperationalIndexes)
+	if !result.Complete {
+		return result, errors.New("schema preflight requirements are not satisfied")
+	}
+	return result, nil
+}