From be71610215b389abfae0161d714fea16a665dbae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=80=97=E5=AD=90?= Date: Fri, 14 Aug 2026 23:55:44 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=98=E5=8C=96=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93=E8=BF=81=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/bootstrap/db.go | 5 +- internal/data/helper.go | 15 ++++++ internal/data/monitor.go | 8 +-- internal/data/scan_event.go | 8 +-- internal/data/tamper.go | 8 +-- internal/data/website_stat.go | 8 +-- internal/migration/migration.go | 6 ++- internal/migration/scan.go | 78 +++++++++++++++++++----------- internal/migration/tamper.go | 5 +- internal/migration/v1.go | 5 +- internal/migration/website_stat.go | 7 ++- 11 files changed, 89 insertions(+), 64 deletions(-) diff --git a/internal/bootstrap/db.go b/internal/bootstrap/db.go index d67d7edb..74e0e2f6 100644 --- a/internal/bootstrap/db.go +++ b/internal/bootstrap/db.go @@ -57,8 +57,5 @@ func NewDB(conf *config.Config) (*gorm.DB, error) { } func NewMigrate(db *gorm.DB) (*gormigrate.Gormigrate, error) { - - return gormigrate.New(db, &gormigrate.Options{ - UseTransaction: true, // Note: MySQL not support DDL transaction - }, migration.Migrations), nil + return gormigrate.New(db, nil, migration.Migrations), nil } diff --git a/internal/data/helper.go b/internal/data/helper.go index bca53ed5..c1714e7d 100644 --- a/internal/data/helper.go +++ b/internal/data/helper.go @@ -79,3 +79,18 @@ func batchUpsert[T any](db *gorm.DB, items []T, conflict clause.OnConflict) erro } return nil } + +// vacuumDB 清理数据后回收文件空间 +func vacuumDB(db *gorm.DB) error { + if err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { + return err + } + if err := db.Exec("VACUUM").Error; err != nil { + return err + } + // 写回 VACUUM 结果并截断文件 + if err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { + return err + } + return db.Exec("PRAGMA optimize").Error +} diff --git a/internal/data/monitor.go b/internal/data/monitor.go index 432777a3..09bebc43 100644 --- a/internal/data/monitor.go +++ b/internal/data/monitor.go @@ -48,11 +48,5 @@ func (r *monitorRepo) List(start, end time.Time) ([]*biz.Monitor, error) { } func (r *monitorRepo) VacuumDB() error { - if err := r.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { - return err - } - if err := r.db.Exec("VACUUM").Error; err != nil { - return err - } - return r.db.Exec("PRAGMA optimize").Error + return vacuumDB(r.db) } diff --git a/internal/data/scan_event.go b/internal/data/scan_event.go index 42dfbc67..1287a539 100644 --- a/internal/data/scan_event.go +++ b/internal/data/scan_event.go @@ -173,13 +173,7 @@ func (r *scanEventRepo) VacuumDB() error { if err := r.db.Exec("DELETE FROM scan_sources WHERE id NOT IN (SELECT source_id FROM scan_events)").Error; err != nil { return err } - if err := r.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { - return err - } - if err := r.db.Exec("VACUUM").Error; err != nil { - return err - } - return r.db.Exec("PRAGMA optimize").Error + return vacuumDB(r.db) } // parseTimeStr 解析 Go time.String() 格式并转为 RFC3339 diff --git a/internal/data/tamper.go b/internal/data/tamper.go index 63391d76..9fcc1d29 100644 --- a/internal/data/tamper.go +++ b/internal/data/tamper.go @@ -89,11 +89,5 @@ func (r *tamperRepo) ClearLogsBefore(t time.Time) error { } func (r *tamperRepo) VacuumDB() error { - if err := r.logDB.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { - return err - } - if err := r.logDB.Exec("VACUUM").Error; err != nil { - return err - } - return r.logDB.Exec("PRAGMA optimize").Error + return vacuumDB(r.logDB) } diff --git a/internal/data/website_stat.go b/internal/data/website_stat.go index f401fe24..266439c5 100644 --- a/internal/data/website_stat.go +++ b/internal/data/website_stat.go @@ -153,13 +153,7 @@ func (r *websiteStatRepo) DeleteBySite(site string) error { } func (r *websiteStatRepo) VacuumDB() error { - if err := r.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { - return err - } - if err := r.db.Exec("VACUUM").Error; err != nil { - return err - } - return r.db.Exec("PRAGMA optimize").Error + return vacuumDB(r.db) } // ========== 蜘蛛统计 ========== diff --git a/internal/migration/migration.go b/internal/migration/migration.go index 3cd72588..019233d3 100644 --- a/internal/migration/migration.go +++ b/internal/migration/migration.go @@ -7,7 +7,7 @@ import ( var Migrations []*gormigrate.Migration -const batchSize = 100 +const batchSize = 1000 func vacuumDB(db *gorm.DB) error { if err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { @@ -16,5 +16,9 @@ func vacuumDB(db *gorm.DB) error { if err := db.Exec("VACUUM").Error; err != nil { return err } + // 写回 VACUUM 结果并截断文件 + if err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE)").Error; err != nil { + return err + } return db.Exec("PRAGMA optimize").Error } diff --git a/internal/migration/scan.go b/internal/migration/scan.go index 8c3939e7..7e2a7f8f 100644 --- a/internal/migration/scan.go +++ b/internal/migration/scan.go @@ -20,50 +20,74 @@ var ScanMigrations = []*gormigrate.Migration{ { ID: "20260814-normalize-scan-events", Migrate: func(tx *gorm.DB) error { - if !tx.Migrator().HasColumn("scan_events", "source_ip") { + legacy := tx.Migrator().HasTable("scan_events_legacy") + if !legacy && !tx.Migrator().HasColumn("scan_events", "source_ip") { return nil } - return tx.Transaction(func(tx *gorm.DB) error { - if err := tx.Migrator().RenameTable("scan_events", "scan_events_legacy"); err != nil { - return err - } - if err := tx.Exec("DROP INDEX IF EXISTS idx_scan_unique").Error; err != nil { - return err - } - if err := tx.Exec("DROP INDEX IF EXISTS idx_scan_date").Error; err != nil { - return err - } - if err := tx.AutoMigrate(&biz.ScanSource{}, &biz.ScanEvent{}); err != nil { + if !legacy { + if err := tx.Transaction(func(tx *gorm.DB) error { + if err := tx.Migrator().RenameTable("scan_events", "scan_events_legacy"); err != nil { + return err + } + if err := tx.Exec("DROP INDEX IF EXISTS idx_scan_unique").Error; err != nil { + return err + } + if err := tx.Exec("DROP INDEX IF EXISTS idx_scan_date").Error; err != nil { + return err + } + return tx.AutoMigrate(&biz.ScanSource{}, &biz.ScanEvent{}) + }); err != nil { return err } + } - if err := tx.Exec(` - INSERT INTO scan_sources (source_ip, country, region, city, isp) - SELECT source_ip, MAX(country), MAX(region), MAX(city), MAX(isp) - FROM scan_events_legacy - WHERE 1 = 1 - GROUP BY source_ip - ON CONFLICT(source_ip) DO UPDATE SET - country = excluded.country, - region = excluded.region, - city = excluded.city, - isp = excluded.isp - `).Error; err != nil { + // 同一 IP 的归属信息取任意一条，扫描来源数量远小于事件数量 + if err := tx.Exec(` + INSERT INTO scan_sources (source_ip, country, region, city, isp) + SELECT source_ip, MAX(country), MAX(region), MAX(city), MAX(isp) + FROM scan_events_legacy + GROUP BY source_ip + ON CONFLICT DO NOTHING + `).Error; err != nil { + return err + } + + // 按主键分批提交 + var cursor uint + if err := tx.Raw("SELECT COALESCE(MAX(id), 0) FROM scan_events").Scan(&cursor).Error; err != nil { + return err + } + for { + var next uint + if err := tx.Raw( + "SELECT COALESCE(MAX(id), 0) FROM (SELECT id FROM scan_events_legacy WHERE id > ? ORDER BY id LIMIT ?)", + cursor, batchSize, + ).Scan(&next).Error; err != nil { return err } + if next == 0 { + break + } if err := tx.Exec(` INSERT INTO scan_events (id, source_id, port, protocol, date, count, first_seen, last_seen) SELECT events.id, sources.id, events.port, events.protocol, events.date, events.count, events.first_seen, events.last_seen FROM scan_events_legacy AS events JOIN scan_sources AS sources ON sources.source_ip = events.source_ip - `).Error; err != nil { + WHERE events.id > ? AND events.id <= ? + ON CONFLICT DO NOTHING + `, cursor, next).Error; err != nil { return err } + cursor = next + } - return tx.Migrator().DropTable("scan_events_legacy") - }) + if err := tx.Migrator().DropTable("scan_events_legacy"); err != nil { + return err + } + + return vacuumDB(tx) }, }, } diff --git a/internal/migration/tamper.go b/internal/migration/tamper.go index a58f44c6..563047cd 100644 --- a/internal/migration/tamper.go +++ b/internal/migration/tamper.go @@ -24,14 +24,17 @@ var TamperMigrations = []*gormigrate.Migration{ } } + var cursor uint for { var items []*biz.TamperLog - if err := tx.Where("typeof(path) = 'text'").Order("id").Limit(batchSize).Find(&items).Error; err != nil { + if err := tx.Where("id > ? AND typeof(path) = 'text'", cursor). + Order("id").Limit(batchSize).Find(&items).Error; err != nil { return err } if len(items) == 0 { break } + cursor = items[len(items)-1].ID if err := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "id"}}, DoUpdates: clause.AssignmentColumns([]string{"path"}), diff --git a/internal/migration/v1.go b/internal/migration/v1.go index d79078fb..143a9402 100644 --- a/internal/migration/v1.go +++ b/internal/migration/v1.go @@ -162,7 +162,10 @@ func init() { if !tx.Migrator().HasTable("monitors") { return nil } - return tx.Migrator().DropTable("monitors") + if err := tx.Migrator().DropTable("monitors"); err != nil { + return err + } + return vacuumDB(tx) }, }) Migrations = append(Migrations, &gormigrate.Migration{ diff --git a/internal/migration/website_stat.go b/internal/migration/website_stat.go index 48f42536..1762bd8f 100644 --- a/internal/migration/website_stat.go +++ b/internal/migration/website_stat.go @@ -22,15 +22,18 @@ var WebsiteStatMigrations = []*gormigrate.Migration{ { ID: "20260814-compress-website-error-logs", Migrate: func(tx *gorm.DB) error { + var cursor uint for { var items []*biz.WebsiteErrorLog - if err := tx.Where("typeof(uri) = 'text' OR typeof(ua) = 'text' OR typeof(body) = 'text'"). - Order("id").Limit(batchSize).Find(&items).Error; err != nil { + if err := tx.Where( + "id > ? AND (typeof(uri) = 'text' OR typeof(ua) = 'text' OR typeof(body) = 'text')", cursor, + ).Order("id").Limit(batchSize).Find(&items).Error; err != nil { return err } if len(items) == 0 { break } + cursor = items[len(items)-1].ID if err := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "id"}}, DoUpdates: clause.AssignmentColumns([]string{"uri", "ua", "body"}),