From 4fa8f073a34a4e1418cf68b42029dc4caa94e2ba Mon Sep 17 00:00:00 2001 From: ryan Date: Fri, 20 Mar 2026 23:29:18 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=8A=9F=E8=83=BD]=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E8=B7=AF=E7=94=B1=E9=A1=B5=E9=9D=A2=EF=BC=8C?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E8=BE=93=E5=85=A5=E7=BB=84=E4=BB=B6=E5=92=8C?= =?UTF-8?q?=E6=A0=B7=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- openflare_server/model/main.go | 649 ----------------- openflare_server/model/migrations.go | 652 ++++++++++++++++++ .../components/proxy-routes-page.tsx | 22 +- 3 files changed, 663 insertions(+), 660 deletions(-) create mode 100644 openflare_server/model/migrations.go diff --git a/openflare_server/model/main.go b/openflare_server/model/main.go index bbefea26..a4c0aaf7 100644 --- a/openflare_server/model/main.go +++ b/openflare_server/model/main.go @@ -1,21 +1,16 @@ package model import ( - "encoding/json" - "errors" "fmt" "github.com/glebarez/sqlite" "gorm.io/driver/postgres" "gorm.io/gorm" "gorm.io/gorm/schema" "log/slog" - "net" - "net/url" "openflare/common" "openflare/utils/security" "os" "reflect" - "strings" "sync" ) @@ -27,13 +22,6 @@ type dbModel struct { hasIDPK bool } -type databaseSchemaMigration struct { - fromVersion int - toVersion int - migrate func(db *gorm.DB, backend string) error - validate func(db *gorm.DB, backend string) error -} - func registeredModels() []any { return []any{ &File{}, @@ -80,16 +68,6 @@ func buildDBModels() ([]dbModel, error) { return result, nil } -func migrateProxyRouteEnableHTTPSColumn(db *gorm.DB) error { - if !db.Migrator().HasTable(&ProxyRoute{}) { - return nil - } - if db.Migrator().HasColumn(&ProxyRoute{}, "enable_https") || !db.Migrator().HasColumn(&ProxyRoute{}, "enable_http_s") { - return nil - } - return db.Migrator().RenameColumn(&ProxyRoute{}, "enable_http_s", "enable_https") -} - func createRootAccountIfNeed() error { var user User //if user.Status != common.UserStatusEnabled { @@ -141,633 +119,6 @@ func autoMigrateAll(db *gorm.DB) error { return nil } -func autoMigrateSchemaMetadata(db *gorm.DB) error { - for _, item := range schemaMetadataModels() { - if err := db.AutoMigrate(item); err != nil { - return err - } - } - return nil -} - -func migrateTextColumns(db *gorm.DB, backend string) error { - if backend != "postgres" { - return nil - } - type textColumn struct { - model any - table string - column string - } - columns := []textColumn{ - {model: &Node{}, table: "nodes", column: "openresty_message"}, - {model: &Node{}, table: "nodes", column: "last_error"}, - {model: &ApplyLog{}, table: "apply_logs", column: "message"}, - {model: &NodeHealthEvent{}, table: "node_health_events", column: "message"}, - } - for _, item := range columns { - if !db.Migrator().HasTable(item.model) || !db.Migrator().HasColumn(item.model, item.column) { - continue - } - sql := fmt.Sprintf(`ALTER TABLE "%s" ALTER COLUMN "%s" TYPE text`, item.table, item.column) - if err := db.Exec(sql).Error; err != nil { - return fmt.Errorf("migrate column %s.%s to text failed: %w", item.table, item.column, err) - } - } - return nil -} - -func migrateObservabilityLegacyColumns(db *gorm.DB) error { - if db == nil { - return nil - } - if !db.Migrator().HasTable(&NodeHealthEvent{}) || !db.Migrator().HasColumn(&NodeHealthEvent{}, "raw_json") { - return nil - } - type legacyHealthEventRaw struct { - ID uint - RawJSON string - MetadataJSON string - } - type legacyHealthEventPayload struct { - Metadata map[string]string `json:"metadata"` - } - - var rows []legacyHealthEventRaw - if err := db.Model(&NodeHealthEvent{}). - Select("id, raw_json, metadata_json"). - Where("raw_json <> '' AND (metadata_json IS NULL OR metadata_json = '')"). - Find(&rows).Error; err != nil { - return fmt.Errorf("query legacy node health event raw_json failed: %w", err) - } - for _, row := range rows { - var payload legacyHealthEventPayload - if err := json.Unmarshal([]byte(row.RawJSON), &payload); err != nil { - continue - } - if len(payload.Metadata) == 0 { - continue - } - metadataJSON, err := json.Marshal(payload.Metadata) - if err != nil { - continue - } - if err := db.Model(&NodeHealthEvent{}). - Where("id = ?", row.ID). - Update("metadata_json", string(metadataJSON)).Error; err != nil { - return fmt.Errorf("migrate node health event metadata_json failed: %w", err) - } - } - return nil -} - -func applyCurrentSchema(db *gorm.DB, backend string) error { - if err := autoMigrateSchemaMetadata(db); err != nil { - return err - } - if err := migrateProxyRouteEnableHTTPSColumn(db); err != nil { - return err - } - if err := autoMigrateAll(db); err != nil { - return err - } - if err := migrateTextColumns(db, backend); err != nil { - return err - } - if err := migrateObservabilityLegacyColumns(db); err != nil { - return err - } - return nil -} - -func loadDatabaseSchemaVersion(db *gorm.DB) (int, bool, error) { - if db == nil { - return 0, false, nil - } - if !db.Migrator().HasTable(&DatabaseSchemaVersion{}) { - return 0, false, nil - } - var state DatabaseSchemaVersion - err := db.Where("id = ?", databaseSchemaVersionRowID).First(&state).Error - if errors.Is(err, gorm.ErrRecordNotFound) { - return 0, false, nil - } - if err != nil { - return 0, false, err - } - return state.Version, true, nil -} - -func saveDatabaseSchemaVersion(db *gorm.DB, version int) error { - return db.Save(&DatabaseSchemaVersion{ - ID: databaseSchemaVersionRowID, - Version: version, - }).Error -} - -func validateDatabaseSchemaV2(db *gorm.DB, backend string) error { - if db == nil { - return fmt.Errorf("database handle is nil") - } - if !db.Migrator().HasTable(&DatabaseSchemaVersion{}) { - return fmt.Errorf("table %s is missing", (&DatabaseSchemaVersion{}).TableName()) - } - models, err := buildDBModels() - if err != nil { - return err - } - for _, item := range models { - if isShardedObservabilityTable(item.tableName) { - for _, table := range observabilityShardTables(item.tableName) { - if !db.Migrator().HasTable(table) { - return fmt.Errorf("sharded table %s is missing", table) - } - } - continue - } - if !db.Migrator().HasTable(item.value) { - return fmt.Errorf("table %s is missing", item.tableName) - } - } - if !db.Migrator().HasColumn(&NodeHealthEvent{}, "metadata_json") { - return fmt.Errorf("column node_health_events.metadata_json is missing") - } - _ = backend - return nil -} - -func validateDatabaseSchemaV3(db *gorm.DB, backend string) error { - if err := validateDatabaseSchemaV2(db, backend); err != nil { - return err - } - for _, baseTable := range shardedObservabilityBaseTables() { - for _, table := range observabilityShardTables(baseTable) { - legacyTable := legacyObservabilityShardTableName(table) - if db.Migrator().HasTable(legacyTable) { - return fmt.Errorf("legacy sharded table %s still exists", legacyTable) - } - } - } - return nil -} - -func validateDatabaseSchemaV4(db *gorm.DB, backend string) error { - if err := validateDatabaseSchemaV3(db, backend); err != nil { - return err - } - if !db.Migrator().HasTable(&Origin{}) { - return fmt.Errorf("table origins is missing") - } - if !db.Migrator().HasColumn(&ProxyRoute{}, "origin_id") { - return fmt.Errorf("column proxy_routes.origin_id is missing") - } - return nil -} - -func renameLegacyObservabilityShardTables(db *gorm.DB) error { - for _, baseTable := range shardedObservabilityBaseTables() { - for _, table := range observabilityShardTables(baseTable) { - legacyTable := legacyObservabilityShardTableName(table) - if db.Migrator().HasTable(legacyTable) { - return fmt.Errorf("legacy sharded table %s already exists", legacyTable) - } - if !db.Migrator().HasTable(table) { - continue - } - if err := db.Migrator().RenameTable(table, legacyTable); err != nil { - return fmt.Errorf("rename sharded table %s to %s failed: %w", table, legacyTable, err) - } - if err := dropLegacyObservabilitySecondaryIndexes(db, legacyTable); err != nil { - return err - } - } - } - return nil -} - -func dropLegacyObservabilitySecondaryIndexes(db *gorm.DB, table string) error { - db = sessionIgnoringSharding(db) - if db == nil { - return fmt.Errorf("database handle is nil") - } - backend := baseDialector(db).Name() - indexes := make([]string, 0) - switch backend { - case "sqlite": - if err := db.Raw( - `SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ? AND name LIKE 'idx_%'`, - table, - ).Scan(&indexes).Error; err != nil { - return fmt.Errorf("list indexes for %s failed: %w", table, err) - } - case "postgres": - if err := db.Raw( - `SELECT indexname FROM pg_indexes WHERE schemaname = current_schema() AND tablename = ? AND indexname LIKE 'idx_%'`, - table, - ).Scan(&indexes).Error; err != nil { - return fmt.Errorf("list indexes for %s failed: %w", table, err) - } - default: - return fmt.Errorf("unsupported database backend %s", backend) - } - for _, indexName := range indexes { - if err := db.Exec(fmt.Sprintf(`DROP INDEX IF EXISTS "%s"`, indexName)).Error; err != nil { - return fmt.Errorf("drop legacy index %s failed: %w", indexName, err) - } - } - return nil -} - -func autoMigrateObservabilityShardTables(db *gorm.DB) error { - db = sessionIgnoringSharding(db) - if db == nil { - return fmt.Errorf("database handle is nil") - } - dialector := baseDialector(db) - if dialector == nil { - return fmt.Errorf("database dialector is nil") - } - type shardedTable struct { - model any - base string - } - tables := []shardedTable{ - {model: &NodeMetricSnapshot{}, base: "node_metric_snapshots"}, - {model: &NodeRequestReport{}, base: "node_request_reports"}, - {model: &NodeAccessLog{}, base: "node_access_logs"}, - } - for _, item := range tables { - for _, table := range observabilityShardTables(item.base) { - tx := db.Table(table) - if err := dialector.Migrator(tx).AutoMigrate(item.model); err != nil { - return fmt.Errorf("auto migrate sharded table %s failed: %w", table, err) - } - } - } - return nil -} - -func dropLegacyObservabilityShardTables(db *gorm.DB) error { - db = sessionIgnoringSharding(db) - if db == nil { - return fmt.Errorf("database handle is nil") - } - for _, baseTable := range shardedObservabilityBaseTables() { - for _, table := range observabilityShardTables(baseTable) { - legacyTable := legacyObservabilityShardTableName(table) - if !db.Migrator().HasTable(legacyTable) { - continue - } - if err := db.Exec(fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, legacyTable)).Error; err != nil { - return fmt.Errorf("drop legacy sharded table %s failed: %w", legacyTable, err) - } - } - } - return nil -} - -func migrateLegacyNodeMetricSnapshots(db *gorm.DB) error { - for _, table := range observabilityShardTables("node_metric_snapshots") { - legacyTable := legacyObservabilityShardTableName(table) - if !db.Migrator().HasTable(legacyTable) { - continue - } - var lastSeenID uint - for { - var rows []NodeMetricSnapshot - query := db.Table(legacyTable).Order("id ASC").Limit(500) - if lastSeenID > 0 { - query = query.Where("id > ?", lastSeenID) - } - if err := query.Find(&rows).Error; err != nil { - return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) - } - if len(rows) == 0 { - break - } - lastSeenID = rows[len(rows)-1].ID - grouped := make(map[string][]NodeMetricSnapshot, observabilityShardCount) - for index := range rows { - rows[index].ID = 0 - if err := assignObservabilityID(&rows[index].ID); err != nil { - return err - } - targetTable := observabilityShardTableForID("node_metric_snapshots", rows[index].ID) - grouped[targetTable] = append(grouped[targetTable], rows[index]) - } - for targetTable, batch := range grouped { - if err := db.Table(targetTable).Create(&batch).Error; err != nil { - return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) - } - } - } - } - return nil -} - -func migrateLegacyNodeRequestReports(db *gorm.DB) error { - for _, table := range observabilityShardTables("node_request_reports") { - legacyTable := legacyObservabilityShardTableName(table) - if !db.Migrator().HasTable(legacyTable) { - continue - } - var lastSeenID uint - for { - var rows []NodeRequestReport - query := db.Table(legacyTable).Order("id ASC").Limit(500) - if lastSeenID > 0 { - query = query.Where("id > ?", lastSeenID) - } - if err := query.Find(&rows).Error; err != nil { - return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) - } - if len(rows) == 0 { - break - } - lastSeenID = rows[len(rows)-1].ID - grouped := make(map[string][]NodeRequestReport, observabilityShardCount) - for index := range rows { - rows[index].ID = 0 - if err := assignObservabilityID(&rows[index].ID); err != nil { - return err - } - targetTable := observabilityShardTableForID("node_request_reports", rows[index].ID) - grouped[targetTable] = append(grouped[targetTable], rows[index]) - } - for targetTable, batch := range grouped { - if err := db.Table(targetTable).Create(&batch).Error; err != nil { - return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) - } - } - } - } - return nil -} - -func migrateLegacyNodeAccessLogs(db *gorm.DB) error { - for _, table := range observabilityShardTables("node_access_logs") { - legacyTable := legacyObservabilityShardTableName(table) - if !db.Migrator().HasTable(legacyTable) { - continue - } - var lastSeenID uint - for { - var rows []NodeAccessLog - query := db.Table(legacyTable).Order("id ASC").Limit(500) - if lastSeenID > 0 { - query = query.Where("id > ?", lastSeenID) - } - if err := query.Find(&rows).Error; err != nil { - return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) - } - if len(rows) == 0 { - break - } - lastSeenID = rows[len(rows)-1].ID - grouped := make(map[string][]NodeAccessLog, observabilityShardCount) - for index := range rows { - rows[index].ID = 0 - if err := assignObservabilityID(&rows[index].ID); err != nil { - return err - } - targetTable := observabilityShardTableForID("node_access_logs", rows[index].ID) - grouped[targetTable] = append(grouped[targetTable], rows[index]) - } - for targetTable, batch := range grouped { - if err := db.Table(targetTable).Create(&batch).Error; err != nil { - return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) - } - } - } - } - return nil -} - -func migrateObservabilityShardsToID(db *gorm.DB, backend string) error { - if db == nil { - return fmt.Errorf("database handle is nil") - } - _ = backend - if err := renameLegacyObservabilityShardTables(db); err != nil { - return err - } - if err := autoMigrateObservabilityShardTables(db); err != nil { - return err - } - if err := migrateLegacyNodeMetricSnapshots(db); err != nil { - return err - } - if err := migrateLegacyNodeRequestReports(db); err != nil { - return err - } - if err := migrateLegacyNodeAccessLogs(db); err != nil { - return err - } - return dropLegacyObservabilityShardTables(db) -} - -func normalizeOriginAddressForMigration(raw string) string { - return strings.ToLower(strings.TrimSpace(raw)) -} - -func extractOriginAddressForMigration(rawURL string) string { - parsed, err := url.ParseRequestURI(strings.TrimSpace(rawURL)) - if err != nil { - return "" - } - return normalizeOriginAddressForMigration(parsed.Hostname()) -} - -func backfillOriginsFromProxyRoutes(db *gorm.DB) error { - if db == nil { - return fmt.Errorf("database handle is nil") - } - if !db.Migrator().HasTable(&Origin{}) || !db.Migrator().HasTable(&ProxyRoute{}) { - return nil - } - - var routes []ProxyRoute - if err := db.Order("id asc").Find(&routes).Error; err != nil { - return fmt.Errorf("list proxy routes for origin backfill failed: %w", err) - } - - type originSeed struct { - ID uint - Address string - } - - originByAddress := make(map[string]originSeed) - var origins []Origin - if err := db.Order("id asc").Find(&origins).Error; err != nil { - return fmt.Errorf("list origins for backfill failed: %w", err) - } - for _, origin := range origins { - address := normalizeOriginAddressForMigration(origin.Address) - if address == "" { - continue - } - originByAddress[address] = originSeed{ID: origin.ID, Address: address} - } - - for _, route := range routes { - address := extractOriginAddressForMigration(route.OriginURL) - if address == "" { - continue - } - origin, ok := originByAddress[address] - if !ok { - name := address - if ip := net.ParseIP(address); ip != nil { - name = ip.String() - } - record := Origin{ - Name: name, - Address: address, - Remark: "", - } - if err := db.Create(&record).Error; err != nil { - return fmt.Errorf("create origin for address %s failed: %w", address, err) - } - origin = originSeed{ID: record.ID, Address: address} - originByAddress[address] = origin - } - if route.OriginID != nil && *route.OriginID == origin.ID { - continue - } - if err := db.Model(&ProxyRoute{}). - Where("id = ?", route.ID). - Update("origin_id", origin.ID).Error; err != nil { - return fmt.Errorf("backfill proxy route %d origin_id failed: %w", route.ID, err) - } - } - - return nil -} - -func migrateOriginsSchema(db *gorm.DB, backend string) error { - if err := applyCurrentSchema(db, backend); err != nil { - return err - } - return backfillOriginsFromProxyRoutes(db) -} - -func databaseSchemaMigrations() []databaseSchemaMigration { - return []databaseSchemaMigration{ - { - fromVersion: 1, - toVersion: 2, - migrate: applyCurrentSchema, - validate: validateDatabaseSchemaV2, - }, - { - fromVersion: 2, - toVersion: 3, - migrate: migrateObservabilityShardsToID, - validate: validateDatabaseSchemaV3, - }, - { - fromVersion: 3, - toVersion: 4, - migrate: migrateOriginsSchema, - validate: validateDatabaseSchemaV4, - }, - } -} - -func databaseSchemaMigrationMap() map[int]databaseSchemaMigration { - migrations := make(map[int]databaseSchemaMigration, len(databaseSchemaMigrations())) - for _, item := range databaseSchemaMigrations() { - migrations[item.fromVersion] = item - } - return migrations -} - -func runDatabaseSchemaMigration(db *gorm.DB, backend string, migration databaseSchemaMigration) error { - if backend == "sqlite" { - if err := migration.migrate(db, backend); err != nil { - return fmt.Errorf("migrate database schema from v%d to v%d failed: %w", migration.fromVersion, migration.toVersion, err) - } - if err := migration.validate(db, backend); err != nil { - return fmt.Errorf("validate database schema v%d failed: %w", migration.toVersion, err) - } - if err := saveDatabaseSchemaVersion(db, migration.toVersion); err != nil { - return fmt.Errorf("persist database schema version v%d failed: %w", migration.toVersion, err) - } - return nil - } - - return db.Transaction(func(tx *gorm.DB) error { - if err := migration.migrate(tx, backend); err != nil { - return fmt.Errorf("migrate database schema from v%d to v%d failed: %w", migration.fromVersion, migration.toVersion, err) - } - if err := migration.validate(tx, backend); err != nil { - return fmt.Errorf("validate database schema v%d failed: %w", migration.toVersion, err) - } - if err := saveDatabaseSchemaVersion(tx, migration.toVersion); err != nil { - return fmt.Errorf("persist database schema version v%d failed: %w", migration.toVersion, err) - } - return nil - }) -} - -func upgradeDatabaseSchema(db *gorm.DB, backend string, version int) error { - if version > currentDatabaseSchemaVersion { - return fmt.Errorf("database schema version %d is newer than application version %d", version, currentDatabaseSchemaVersion) - } - if version == currentDatabaseSchemaVersion { - return nil - } - migrationMap := databaseSchemaMigrationMap() - for version < currentDatabaseSchemaVersion { - migration, ok := migrationMap[version] - if !ok { - return fmt.Errorf("database schema migration from v%d is not defined", version) - } - if err := runDatabaseSchemaMigration(db, backend, migration); err != nil { - return err - } - version = migration.toVersion - } - return nil -} - -func initializeFreshDatabaseSchema(db *gorm.DB, backend string) error { - if err := applyCurrentSchema(db, backend); err != nil { - return err - } - if err := backfillOriginsFromProxyRoutes(db); err != nil { - return err - } - if err := migrateSQLiteDataIfNeeded(db, backend); err != nil { - return err - } - if err := validateDatabaseSchemaV4(db, backend); err != nil { - return err - } - return saveDatabaseSchemaVersion(db, currentDatabaseSchemaVersion) -} - -func ensureDatabaseSchemaUpToDate(db *gorm.DB, backend string) error { - version, exists, err := loadDatabaseSchemaVersion(db) - if err != nil { - return err - } - if exists { - return upgradeDatabaseSchema(db, backend, version) - } - empty, err := isDatabaseEmpty(db) - if err != nil { - return err - } - if empty { - return initializeFreshDatabaseSchema(db, backend) - } - if err := autoMigrateSchemaMetadata(db); err != nil { - return err - } - return upgradeDatabaseSchema(db, backend, legacyDatabaseSchemaVersion) -} - func isDatabaseEmpty(db *gorm.DB) (bool, error) { models, err := buildDBModels() if err != nil { diff --git a/openflare_server/model/migrations.go b/openflare_server/model/migrations.go new file mode 100644 index 00000000..d83c3926 --- /dev/null +++ b/openflare_server/model/migrations.go @@ -0,0 +1,652 @@ +package model + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "net/url" + "strings" + + "gorm.io/gorm" +) + +type databaseSchemaMigration struct { + fromVersion int + toVersion int + migrate func(db *gorm.DB, backend string) error + validate func(db *gorm.DB, backend string) error +} + +func autoMigrateSchemaMetadata(db *gorm.DB) error { + for _, item := range schemaMetadataModels() { + if err := db.AutoMigrate(item); err != nil { + return err + } + } + return nil +} + +func migrateProxyRouteEnableHTTPSColumn(db *gorm.DB) error { + if !db.Migrator().HasTable(&ProxyRoute{}) { + return nil + } + if db.Migrator().HasColumn(&ProxyRoute{}, "enable_https") || !db.Migrator().HasColumn(&ProxyRoute{}, "enable_http_s") { + return nil + } + return db.Migrator().RenameColumn(&ProxyRoute{}, "enable_http_s", "enable_https") +} + +func migrateTextColumns(db *gorm.DB, backend string) error { + if backend != "postgres" { + return nil + } + type textColumn struct { + model any + table string + column string + } + columns := []textColumn{ + {model: &Node{}, table: "nodes", column: "openresty_message"}, + {model: &Node{}, table: "nodes", column: "last_error"}, + {model: &ApplyLog{}, table: "apply_logs", column: "message"}, + {model: &NodeHealthEvent{}, table: "node_health_events", column: "message"}, + } + for _, item := range columns { + if !db.Migrator().HasTable(item.model) || !db.Migrator().HasColumn(item.model, item.column) { + continue + } + sql := fmt.Sprintf(`ALTER TABLE "%s" ALTER COLUMN "%s" TYPE text`, item.table, item.column) + if err := db.Exec(sql).Error; err != nil { + return fmt.Errorf("migrate column %s.%s to text failed: %w", item.table, item.column, err) + } + } + return nil +} + +func migrateObservabilityLegacyColumns(db *gorm.DB) error { + if db == nil { + return nil + } + if !db.Migrator().HasTable(&NodeHealthEvent{}) || !db.Migrator().HasColumn(&NodeHealthEvent{}, "raw_json") { + return nil + } + type legacyHealthEventRaw struct { + ID uint + RawJSON string + MetadataJSON string + } + type legacyHealthEventPayload struct { + Metadata map[string]string `json:"metadata"` + } + + var rows []legacyHealthEventRaw + if err := db.Model(&NodeHealthEvent{}). + Select("id, raw_json, metadata_json"). + Where("raw_json <> '' AND (metadata_json IS NULL OR metadata_json = '')"). + Find(&rows).Error; err != nil { + return fmt.Errorf("query legacy node health event raw_json failed: %w", err) + } + for _, row := range rows { + var payload legacyHealthEventPayload + if err := json.Unmarshal([]byte(row.RawJSON), &payload); err != nil { + continue + } + if len(payload.Metadata) == 0 { + continue + } + metadataJSON, err := json.Marshal(payload.Metadata) + if err != nil { + continue + } + if err := db.Model(&NodeHealthEvent{}). + Where("id = ?", row.ID). + Update("metadata_json", string(metadataJSON)).Error; err != nil { + return fmt.Errorf("migrate node health event metadata_json failed: %w", err) + } + } + return nil +} + +func applyCurrentSchema(db *gorm.DB, backend string) error { + if err := autoMigrateSchemaMetadata(db); err != nil { + return err + } + if err := migrateProxyRouteEnableHTTPSColumn(db); err != nil { + return err + } + if err := autoMigrateAll(db); err != nil { + return err + } + if err := migrateTextColumns(db, backend); err != nil { + return err + } + if err := migrateObservabilityLegacyColumns(db); err != nil { + return err + } + return nil +} + +func loadDatabaseSchemaVersion(db *gorm.DB) (int, bool, error) { + if db == nil { + return 0, false, nil + } + if !db.Migrator().HasTable(&DatabaseSchemaVersion{}) { + return 0, false, nil + } + var state DatabaseSchemaVersion + err := db.Where("id = ?", databaseSchemaVersionRowID).First(&state).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return 0, false, nil + } + if err != nil { + return 0, false, err + } + return state.Version, true, nil +} + +func saveDatabaseSchemaVersion(db *gorm.DB, version int) error { + return db.Save(&DatabaseSchemaVersion{ + ID: databaseSchemaVersionRowID, + Version: version, + }).Error +} + +func validateDatabaseSchemaV2(db *gorm.DB, backend string) error { + if db == nil { + return fmt.Errorf("database handle is nil") + } + if !db.Migrator().HasTable(&DatabaseSchemaVersion{}) { + return fmt.Errorf("table %s is missing", (&DatabaseSchemaVersion{}).TableName()) + } + models, err := buildDBModels() + if err != nil { + return err + } + for _, item := range models { + if isShardedObservabilityTable(item.tableName) { + for _, table := range observabilityShardTables(item.tableName) { + if !db.Migrator().HasTable(table) { + return fmt.Errorf("sharded table %s is missing", table) + } + } + continue + } + if !db.Migrator().HasTable(item.value) { + return fmt.Errorf("table %s is missing", item.tableName) + } + } + if !db.Migrator().HasColumn(&NodeHealthEvent{}, "metadata_json") { + return fmt.Errorf("column node_health_events.metadata_json is missing") + } + _ = backend + return nil +} + +func validateDatabaseSchemaV3(db *gorm.DB, backend string) error { + if err := validateDatabaseSchemaV2(db, backend); err != nil { + return err + } + for _, baseTable := range shardedObservabilityBaseTables() { + for _, table := range observabilityShardTables(baseTable) { + legacyTable := legacyObservabilityShardTableName(table) + if db.Migrator().HasTable(legacyTable) { + return fmt.Errorf("legacy sharded table %s still exists", legacyTable) + } + } + } + return nil +} + +func validateDatabaseSchemaV4(db *gorm.DB, backend string) error { + if err := validateDatabaseSchemaV3(db, backend); err != nil { + return err + } + if !db.Migrator().HasTable(&Origin{}) { + return fmt.Errorf("table origins is missing") + } + if !db.Migrator().HasColumn(&ProxyRoute{}, "origin_id") { + return fmt.Errorf("column proxy_routes.origin_id is missing") + } + return nil +} + +func renameLegacyObservabilityShardTables(db *gorm.DB) error { + for _, baseTable := range shardedObservabilityBaseTables() { + for _, table := range observabilityShardTables(baseTable) { + legacyTable := legacyObservabilityShardTableName(table) + if db.Migrator().HasTable(legacyTable) { + return fmt.Errorf("legacy sharded table %s already exists", legacyTable) + } + if !db.Migrator().HasTable(table) { + continue + } + if err := db.Migrator().RenameTable(table, legacyTable); err != nil { + return fmt.Errorf("rename sharded table %s to %s failed: %w", table, legacyTable, err) + } + if err := dropLegacyObservabilitySecondaryIndexes(db, legacyTable); err != nil { + return err + } + } + } + return nil +} + +func dropLegacyObservabilitySecondaryIndexes(db *gorm.DB, table string) error { + db = sessionIgnoringSharding(db) + if db == nil { + return fmt.Errorf("database handle is nil") + } + backend := baseDialector(db).Name() + indexes := make([]string, 0) + switch backend { + case "sqlite": + if err := db.Raw( + `SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ? AND name LIKE 'idx_%'`, + table, + ).Scan(&indexes).Error; err != nil { + return fmt.Errorf("list indexes for %s failed: %w", table, err) + } + case "postgres": + if err := db.Raw( + `SELECT indexname FROM pg_indexes WHERE schemaname = current_schema() AND tablename = ? AND indexname LIKE 'idx_%'`, + table, + ).Scan(&indexes).Error; err != nil { + return fmt.Errorf("list indexes for %s failed: %w", table, err) + } + default: + return fmt.Errorf("unsupported database backend %s", backend) + } + for _, indexName := range indexes { + if err := db.Exec(fmt.Sprintf(`DROP INDEX IF EXISTS "%s"`, indexName)).Error; err != nil { + return fmt.Errorf("drop legacy index %s failed: %w", indexName, err) + } + } + return nil +} + +func autoMigrateObservabilityShardTables(db *gorm.DB) error { + db = sessionIgnoringSharding(db) + if db == nil { + return fmt.Errorf("database handle is nil") + } + dialector := baseDialector(db) + if dialector == nil { + return fmt.Errorf("database dialector is nil") + } + type shardedTable struct { + model any + base string + } + tables := []shardedTable{ + {model: &NodeMetricSnapshot{}, base: "node_metric_snapshots"}, + {model: &NodeRequestReport{}, base: "node_request_reports"}, + {model: &NodeAccessLog{}, base: "node_access_logs"}, + } + for _, item := range tables { + for _, table := range observabilityShardTables(item.base) { + tx := db.Table(table) + if err := dialector.Migrator(tx).AutoMigrate(item.model); err != nil { + return fmt.Errorf("auto migrate sharded table %s failed: %w", table, err) + } + } + } + return nil +} + +func dropLegacyObservabilityShardTables(db *gorm.DB) error { + db = sessionIgnoringSharding(db) + if db == nil { + return fmt.Errorf("database handle is nil") + } + for _, baseTable := range shardedObservabilityBaseTables() { + for _, table := range observabilityShardTables(baseTable) { + legacyTable := legacyObservabilityShardTableName(table) + if !db.Migrator().HasTable(legacyTable) { + continue + } + if err := db.Exec(fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, legacyTable)).Error; err != nil { + return fmt.Errorf("drop legacy sharded table %s failed: %w", legacyTable, err) + } + } + } + return nil +} + +func migrateLegacyNodeMetricSnapshots(db *gorm.DB) error { + for _, table := range observabilityShardTables("node_metric_snapshots") { + legacyTable := legacyObservabilityShardTableName(table) + if !db.Migrator().HasTable(legacyTable) { + continue + } + var lastSeenID uint + for { + var rows []NodeMetricSnapshot + query := db.Table(legacyTable).Order("id ASC").Limit(500) + if lastSeenID > 0 { + query = query.Where("id > ?", lastSeenID) + } + if err := query.Find(&rows).Error; err != nil { + return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) + } + if len(rows) == 0 { + break + } + lastSeenID = rows[len(rows)-1].ID + grouped := make(map[string][]NodeMetricSnapshot, observabilityShardCount) + for index := range rows { + rows[index].ID = 0 + if err := assignObservabilityID(&rows[index].ID); err != nil { + return err + } + targetTable := observabilityShardTableForID("node_metric_snapshots", rows[index].ID) + grouped[targetTable] = append(grouped[targetTable], rows[index]) + } + for targetTable, batch := range grouped { + if err := db.Table(targetTable).Create(&batch).Error; err != nil { + return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) + } + } + } + } + return nil +} + +func migrateLegacyNodeRequestReports(db *gorm.DB) error { + for _, table := range observabilityShardTables("node_request_reports") { + legacyTable := legacyObservabilityShardTableName(table) + if !db.Migrator().HasTable(legacyTable) { + continue + } + var lastSeenID uint + for { + var rows []NodeRequestReport + query := db.Table(legacyTable).Order("id ASC").Limit(500) + if lastSeenID > 0 { + query = query.Where("id > ?", lastSeenID) + } + if err := query.Find(&rows).Error; err != nil { + return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) + } + if len(rows) == 0 { + break + } + lastSeenID = rows[len(rows)-1].ID + grouped := make(map[string][]NodeRequestReport, observabilityShardCount) + for index := range rows { + rows[index].ID = 0 + if err := assignObservabilityID(&rows[index].ID); err != nil { + return err + } + targetTable := observabilityShardTableForID("node_request_reports", rows[index].ID) + grouped[targetTable] = append(grouped[targetTable], rows[index]) + } + for targetTable, batch := range grouped { + if err := db.Table(targetTable).Create(&batch).Error; err != nil { + return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) + } + } + } + } + return nil +} + +func migrateLegacyNodeAccessLogs(db *gorm.DB) error { + for _, table := range observabilityShardTables("node_access_logs") { + legacyTable := legacyObservabilityShardTableName(table) + if !db.Migrator().HasTable(legacyTable) { + continue + } + var lastSeenID uint + for { + var rows []NodeAccessLog + query := db.Table(legacyTable).Order("id ASC").Limit(500) + if lastSeenID > 0 { + query = query.Where("id > ?", lastSeenID) + } + if err := query.Find(&rows).Error; err != nil { + return fmt.Errorf("query legacy sharded table %s failed: %w", legacyTable, err) + } + if len(rows) == 0 { + break + } + lastSeenID = rows[len(rows)-1].ID + grouped := make(map[string][]NodeAccessLog, observabilityShardCount) + for index := range rows { + rows[index].ID = 0 + if err := assignObservabilityID(&rows[index].ID); err != nil { + return err + } + targetTable := observabilityShardTableForID("node_access_logs", rows[index].ID) + grouped[targetTable] = append(grouped[targetTable], rows[index]) + } + for targetTable, batch := range grouped { + if err := db.Table(targetTable).Create(&batch).Error; err != nil { + return fmt.Errorf("write migrated rows into %s failed: %w", targetTable, err) + } + } + } + } + return nil +} + +func normalizeOriginAddressForMigration(raw string) string { + return strings.ToLower(strings.TrimSpace(raw)) +} + +func extractOriginAddressForMigration(rawURL string) string { + parsed, err := url.ParseRequestURI(strings.TrimSpace(rawURL)) + if err != nil { + return "" + } + return normalizeOriginAddressForMigration(parsed.Hostname()) +} + +func backfillOriginsFromProxyRoutes(db *gorm.DB) error { + if db == nil { + return fmt.Errorf("database handle is nil") + } + if !db.Migrator().HasTable(&Origin{}) || !db.Migrator().HasTable(&ProxyRoute{}) { + return nil + } + + var routes []ProxyRoute + if err := db.Order("id asc").Find(&routes).Error; err != nil { + return fmt.Errorf("list proxy routes for origin backfill failed: %w", err) + } + + type originSeed struct { + ID uint + Address string + } + + originByAddress := make(map[string]originSeed) + var origins []Origin + if err := db.Order("id asc").Find(&origins).Error; err != nil { + return fmt.Errorf("list origins for backfill failed: %w", err) + } + for _, origin := range origins { + address := normalizeOriginAddressForMigration(origin.Address) + if address == "" { + continue + } + originByAddress[address] = originSeed{ID: origin.ID, Address: address} + } + + for _, route := range routes { + address := extractOriginAddressForMigration(route.OriginURL) + if address == "" { + continue + } + origin, ok := originByAddress[address] + if !ok { + name := address + if ip := net.ParseIP(address); ip != nil { + name = ip.String() + } + record := Origin{ + Name: name, + Address: address, + Remark: "", + } + if err := db.Create(&record).Error; err != nil { + return fmt.Errorf("create origin for address %s failed: %w", address, err) + } + origin = originSeed{ID: record.ID, Address: address} + originByAddress[address] = origin + } + if route.OriginID != nil && *route.OriginID == origin.ID { + continue + } + if err := db.Model(&ProxyRoute{}). + Where("id = ?", route.ID). + Update("origin_id", origin.ID).Error; err != nil { + return fmt.Errorf("backfill proxy route %d origin_id failed: %w", route.ID, err) + } + } + + return nil +} + +// migrateV2 upgrades the legacy schema to the first versioned schema by +// creating schema metadata, applying the current tables, and backfilling +// compatibility columns. +func migrateV2(db *gorm.DB, backend string) error { + return applyCurrentSchema(db, backend) +} + +// migrateV3 upgrades observability shard tables from legacy ID layout to the +// current ID-sharded layout and migrates existing shard data into the new tables. +func migrateV3(db *gorm.DB, backend string) error { + if db == nil { + return fmt.Errorf("database handle is nil") + } + _ = backend + if err := renameLegacyObservabilityShardTables(db); err != nil { + return err + } + if err := autoMigrateObservabilityShardTables(db); err != nil { + return err + } + if err := migrateLegacyNodeMetricSnapshots(db); err != nil { + return err + } + if err := migrateLegacyNodeRequestReports(db); err != nil { + return err + } + if err := migrateLegacyNodeAccessLogs(db); err != nil { + return err + } + return dropLegacyObservabilityShardTables(db) +} + +// migrateV4 introduces the origins schema and backfills proxy route origin +// references from existing origin_url values. +func migrateV4(db *gorm.DB, backend string) error { + if err := applyCurrentSchema(db, backend); err != nil { + return err + } + return backfillOriginsFromProxyRoutes(db) +} + +func databaseSchemaMigrations() []databaseSchemaMigration { + return []databaseSchemaMigration{ + {fromVersion: 1, toVersion: 2, migrate: migrateV2, validate: validateDatabaseSchemaV2}, + {fromVersion: 2, toVersion: 3, migrate: migrateV3, validate: validateDatabaseSchemaV3}, + {fromVersion: 3, toVersion: 4, migrate: migrateV4, validate: validateDatabaseSchemaV4}, + } +} + +func databaseSchemaMigrationMap() map[int]databaseSchemaMigration { + migrations := make(map[int]databaseSchemaMigration, len(databaseSchemaMigrations())) + for _, item := range databaseSchemaMigrations() { + migrations[item.fromVersion] = item + } + return migrations +} + +func runDatabaseSchemaMigration(db *gorm.DB, backend string, migration databaseSchemaMigration) error { + if backend == "sqlite" { + if err := migration.migrate(db, backend); err != nil { + return fmt.Errorf("migrate database schema from v%d to v%d failed: %w", migration.fromVersion, migration.toVersion, err) + } + if err := migration.validate(db, backend); err != nil { + return fmt.Errorf("validate database schema v%d failed: %w", migration.toVersion, err) + } + if err := saveDatabaseSchemaVersion(db, migration.toVersion); err != nil { + return fmt.Errorf("persist database schema version v%d failed: %w", migration.toVersion, err) + } + return nil + } + + return db.Transaction(func(tx *gorm.DB) error { + if err := migration.migrate(tx, backend); err != nil { + return fmt.Errorf("migrate database schema from v%d to v%d failed: %w", migration.fromVersion, migration.toVersion, err) + } + if err := migration.validate(tx, backend); err != nil { + return fmt.Errorf("validate database schema v%d failed: %w", migration.toVersion, err) + } + if err := saveDatabaseSchemaVersion(tx, migration.toVersion); err != nil { + return fmt.Errorf("persist database schema version v%d failed: %w", migration.toVersion, err) + } + return nil + }) +} + +func upgradeDatabaseSchema(db *gorm.DB, backend string, version int) error { + if version > currentDatabaseSchemaVersion { + return fmt.Errorf("database schema version %d is newer than application version %d", version, currentDatabaseSchemaVersion) + } + if version == currentDatabaseSchemaVersion { + return nil + } + migrationMap := databaseSchemaMigrationMap() + for version < currentDatabaseSchemaVersion { + migration, ok := migrationMap[version] + if !ok { + return fmt.Errorf("database schema migration from v%d is not defined", version) + } + if err := runDatabaseSchemaMigration(db, backend, migration); err != nil { + return err + } + version = migration.toVersion + } + return nil +} + +func initializeFreshDatabaseSchema(db *gorm.DB, backend string) error { + if err := applyCurrentSchema(db, backend); err != nil { + return err + } + if err := backfillOriginsFromProxyRoutes(db); err != nil { + return err + } + if err := migrateSQLiteDataIfNeeded(db, backend); err != nil { + return err + } + if err := validateDatabaseSchemaV4(db, backend); err != nil { + return err + } + return saveDatabaseSchemaVersion(db, currentDatabaseSchemaVersion) +} + +func ensureDatabaseSchemaUpToDate(db *gorm.DB, backend string) error { + version, exists, err := loadDatabaseSchemaVersion(db) + if err != nil { + return err + } + if exists { + return upgradeDatabaseSchema(db, backend, version) + } + empty, err := isDatabaseEmpty(db) + if err != nil { + return err + } + if empty { + return initializeFreshDatabaseSchema(db, backend) + } + if err := autoMigrateSchemaMetadata(db); err != nil { + return err + } + return upgradeDatabaseSchema(db, backend, legacyDatabaseSchemaVersion) +} diff --git a/openflare_server/web/features/proxy-routes/components/proxy-routes-page.tsx b/openflare_server/web/features/proxy-routes/components/proxy-routes-page.tsx index d8b20b99..af0e12ab 100644 --- a/openflare_server/web/features/proxy-routes/components/proxy-routes-page.tsx +++ b/openflare_server/web/features/proxy-routes/components/proxy-routes-page.tsx @@ -1848,20 +1848,20 @@ export function ProxyRoutesPage() { ?.message}

)} - - {index === 0 ? ( -
- Origin Path / Query - -
- ) : null} ); })} + + + +