[优化] 界面优化

[功能] 添加数据库观测数据清理功能，支持手动和自动清理策略
[功能] 添加遗留观察性索引和表的删除逻辑，优化数据库迁移过程
2026-09-28 21:56:36 +08:00 · 2026-03-19 21:00:43 +08:00 · 2026-03-19 20:48:45 +08:00 · 2026-03-19 20:26:18 +08:00 · 2026-03-19 17:57:30 +08:00 · 2026-03-19 16:45:22 +08:00
31 changed files with 2663 additions and 213 deletions
@@ -46,4 +46,5 @@ go.work.sum
 *.log

 .DS_Store
-.codex-cache
+.codex-cache
+/.gomodcache/
@@ -57,12 +57,19 @@ go run . --port 3000 --log-dir ./logs
 | `NodeOfflineThreshold` | 节点离线阈值（毫秒） | `120000` |
 | `AgentUpdateRepo` | Agent 自更新仓库 | `Rain-kl/OpenFlare` |
 | `GeoIPProvider` | 节点/IP 归属解析方式 | `ipinfo` |
+| `DatabaseAutoCleanupEnabled` | 是否启用每日自动清理观测数据 | `false` |
+| `DatabaseAutoCleanupRetentionDays` | 自动清理保留天数（至少 1 天） | `30` |
 | `GlobalApiRateLimitNum` / `GlobalApiRateLimitDuration` | 全局 API 限流次数 / 时间窗口 | `300` / `180` |
 | `GlobalWebRateLimitNum` / `GlobalWebRateLimitDuration` | 全局 Web 限流次数 / 时间窗口 | `300` / `180` |
 | `UploadRateLimitNum` / `UploadRateLimitDuration` | 上传接口限流次数 / 时间窗口 | `50` / `60` |
 | `DownloadRateLimitNum` / `DownloadRateLimitDuration` | 下载接口限流次数 / 时间窗口 | `50` / `60` |
 | `CriticalRateLimitNum` / `CriticalRateLimitDuration` | 敏感接口限流次数 / 时间窗口 | `100` / `1200` |

+说明：
+
+* `DatabaseAutoCleanupEnabled` 开启后，Server 会在每天凌晨 3 点自动清理 `node_access_logs`、`node_metric_snapshots`、`node_request_reports` 三类观测数据
+* `DatabaseAutoCleanupRetentionDays` 为统一保留天数，必须大于等于 1；管理端支持手动清理时留空保留天数，以直接删除对应数据集的全部历史记录
+
 ### 1.4 OpenResty 参数

 OpenResty 性能参数与缓存参数继续统一保存在 `Option` 表。当前常用项包括：
@@ -126,6 +126,18 @@
 * 快照与聚合结果采用追加式模型，不覆盖历史
 * 原始访问明细必须有受控保留策略

+### 3.1 数据库版本与迁移
+
+* 任何涉及表结构、索引、列类型、分表规则或内部持久化元数据的修改，都必须同步提升数据库版本号
+* 数据库版本号定义在 `openflare_server/model`，不得只依赖 `AutoMigrate` 隐式升级存量数据库
+* 每次提升数据库版本号时，必须补充从上一版本升级到新版本的显式迁移方法
+* 迁移方法必须包含升级后的校验逻辑；只有校验通过，才能写入新的数据库版本记录
+* 新包启动后必须先检查数据库当前版本，再按顺序逐步升级到目标版本；禁止跳过中间升级步骤直接写目标版本
+* 空库初始化可以直接建立当前版本结构，但初始化完成后仍必须执行同版本校验，并落库当前数据库版本
+* 数据库版本元数据属于内部控制信息，必须保存在独立内部表中，不能混入业务配置表
+* 如果迁移失败或校验失败，启动流程必须中止，且不得提升数据库版本记录
+* 涉及数据库版本变更的提交，必须补充对应的迁移测试或等效回归测试
+
 ## 4. API 与鉴权规范

 ### 4.1 API
@@ -327,18 +327,28 @@ type trafficCountItem struct {
 	value int64
 }

+const accessLogPathMaxRunes = 100
+
 func normalizeAccessLogPath(value string) string {
 	trimmed := strings.TrimSpace(value)
 	if trimmed == "" {
 		return ""
 	}
 	if strings.HasPrefix(trimmed, "http://") || strings.HasPrefix(trimmed, "https://") {
-		return trimmed
+		return truncateAccessLogPath(trimmed)
 	}
 	if strings.HasPrefix(trimmed, "/") {
-		return trimmed
+		return truncateAccessLogPath(trimmed)
 	}
-	return "/" + trimmed
+	return truncateAccessLogPath("/" + trimmed)
+}
+
+func truncateAccessLogPath(value string) string {
+	runes := []rune(value)
+	if len(runes) <= accessLogPathMaxRunes {
+		return value
+	}
+	return string(runes[:accessLogPathMaxRunes])
 }

 func topCounts(values map[string]int64, limit int) map[string]int64 {
@@ -3,6 +3,7 @@ package observability
 import (
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"

 	"openflare-agent/internal/config"
@@ -108,6 +109,31 @@ func TestBuildTrafficObservabilityReturnsAccessLogs(t *testing.T) {
 	}
 }

+func TestBuildTrafficObservabilityTruncatesLongAccessLogPath(t *testing.T) {
+	tempDir := t.TempDir()
+	routeConfigPath := filepath.Join(tempDir, "conf.d", "openflare_routes.conf")
+	if err := os.MkdirAll(filepath.Dir(routeConfigPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+	logPath := filepath.Join(filepath.Dir(routeConfigPath), "openflare_access.log")
+	longPath := "/" + strings.Repeat("a", 140)
+	content := []byte(
+		"{\"ts\":\"2026-03-14T08:00:00Z\",\"host\":\"app.example.com\",\"path\":\"" + longPath + "\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n",
+	)
+	if err := os.WriteFile(logPath, content, 0o644); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+
+	stateStore := state.NewStore(filepath.Join(tempDir, "state.json"))
+	_, accessLogs, _ := BuildTrafficObservability(&config.Config{RouteConfigPath: routeConfigPath}, stateStore, nil)
+	if len(accessLogs) != 1 {
+		t.Fatalf("expected one access log, got %+v", accessLogs)
+	}
+	if got := len([]rune(accessLogs[0].Path)); got != accessLogPathMaxRunes {
+		t.Fatalf("expected truncated path length %d, got %d (%q)", accessLogPathMaxRunes, got, accessLogs[0].Path)
+	}
+}
+
 func TestBuildTrafficReportParsesCombinedAccessLog(t *testing.T) {
 	tempDir := t.TempDir()
 	routeConfigPath := filepath.Join(tempDir, "conf.d", "openflare_routes.conf")
@@ -55,6 +55,8 @@ var NodeOfflineThreshold = 2 * time.Minute
 var AgentHeartbeatInterval = 10000 // milliseconds
 var AgentUpdateRepo = "Rain-kl/OpenFlare"
 var GeoIPProvider = "ipinfo"
+var DatabaseAutoCleanupEnabled = false
+var DatabaseAutoCleanupRetentionDays = 30

 // V5 OpenResty performance settings (hot-reloadable via Option table)
 var OpenRestyWorkerProcesses = "auto"
@@ -0,0 +1,34 @@
+package controller
+
+import (
+	"net/http"
+	"openflare/service"
+
+	"github.com/gin-gonic/gin"
+)
+
+// CleanupDatabaseObservability godoc
+// @Summary Cleanup observability tables
+// @Tags Options
+// @Accept json
+// @Produce json
+// @Security BearerAuth
+// @Success 200 {object} map[string]interface{}
+// @Router /api/option/database/cleanup [post]
+func CleanupDatabaseObservability(c *gin.Context) {
+	var input service.DatabaseCleanupInput
+	if err := decodeOptionalJSONBody(c.Request.Body, &input); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{
+			"success": false,
+			"message": "参数错误",
+			"error":   err.Error(),
+		})
+		return
+	}
+	result, err := service.CleanupDatabaseObservability(input)
+	if err != nil {
+		respondFailure(c, err.Error())
+		return
+	}
+	respondSuccess(c, result)
+}
@@ -73,6 +73,21 @@ func validateGeoIPOption(key string, value string) error {
 	return nil
 }

+func validateDatabaseCleanupOption(key string, value string) error {
+	switch key {
+	case "DatabaseAutoCleanupEnabled":
+		return validateBooleanOption(key, value)
+	case "DatabaseAutoCleanupRetentionDays":
+		intValue, err := strconv.Atoi(value)
+		if err != nil || intValue < 1 {
+			return fmt.Errorf("%s 必须为大于等于 1 的整数天", key)
+		}
+		return nil
+	default:
+		return nil
+	}
+}
+
 func validateOpenRestyOption(key string, value string) error {
 	trimmed := strings.TrimSpace(value)

@@ -271,6 +286,13 @@ func UpdateOption(c *gin.Context) {
 		})
 		return
 	}
+	if err = validateDatabaseCleanupOption(option.Key, option.Value); err != nil {
+		c.JSON(http.StatusOK, gin.H{
+			"success": false,
+			"message": err.Error(),
+		})
+		return
+	}
 	err = model.UpdateOption(option.Key, option.Value)
 	if err != nil {
 		c.JSON(http.StatusOK, gin.H{
@@ -28,6 +28,7 @@ require (
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff // indirect
+	github.com/bwmarrin/snowflake v0.3.0 // indirect
 	github.com/bytedance/sonic v1.11.2 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/chenzhuoyu/base64x v0.0.0-20230717121745-296ad89f973d // indirect
@@ -58,6 +59,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
+	github.com/longbridgeapp/sqlparser v0.3.1 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -67,6 +69,7 @@ require (
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
 	golang.org/x/arch v0.7.0 // indirect
+	golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 // indirect
 	golang.org/x/sync v0.18.0 // indirect
 	golang.org/x/sys v0.38.0 // indirect
 	golang.org/x/text v0.31.0 // indirect
@@ -74,6 +77,7 @@ require (
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	gorm.io/sharding v0.6.2 // indirect
 	modernc.org/libc v1.22.5 // indirect
 	modernc.org/mathutil v1.5.0 // indirect
 	modernc.org/memory v1.5.0 // indirect
@@ -6,6 +6,8 @@ github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff h1:RmdPFa+slIr4SCBg4st/l/vZWVe9QJKMXGO60Bxbe04=
 github.com/boj/redistore v0.0.0-20180917114910-cd5dcc76aeff/go.mod h1:+RTT1BOk5P97fT2CiHkbFQwkK3mjsFAP6zCYV2aXtjw=
+github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0=
+github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE=
 github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
 github.com/bytedance/sonic v1.10.0-rc/go.mod h1:ElCzW+ufi8qKqNW0FY314xriJhyJhuoJ3gFZdAHF7NM=
 github.com/bytedance/sonic v1.11.2 h1:ywfwo0a/3j9HR8wsYGWsIWl2mvRsI950HyoxiBERw5A=
@@ -76,6 +78,7 @@ github.com/go-playground/validator/v10 v10.19.0 h1:ol+5Fu+cSq9JD7SoSqe04GMI92cbn
 github.com/go-playground/validator/v10 v10.19.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM=
 github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
 github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/go-test/deep v1.0.7/go.mod h1:QV8Hv/iy04NyLBxAdO9njL0iVPN1S4d/A3NVv1V36o8=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
 github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
@@ -126,6 +129,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
 github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
 github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI=
+github.com/longbridgeapp/sqlparser v0.3.1 h1:iWOZWGIFgQrJRgobLXUNJdvqGRpbVXkyKUKUA5CNJBE=
+github.com/longbridgeapp/sqlparser v0.3.1/go.mod h1:GIHaUq8zvYyHLCLMJJykx1CdM6LHtkUih/QaJXySSx4=
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.7.6 h1:8yTIVnZgCoiM1TgqoeTl+LfU5Jg6/xL3QhGQnimLYnA=
@@ -157,6 +162,7 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -190,6 +196,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
 golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
+golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 h1:m64FZMko/V45gv0bNmrNYoDEq8U5YUhetc9cBWKS1TQ=
+golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63/go.mod h1:0v4NqG35kSWCMzLaMeX+IQrlSnVE/bqGSyC2cz/9Le8=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA=
 golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w=
@@ -253,6 +261,8 @@ gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4=
 gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo=
 gorm.io/gorm v1.25.10 h1:dQpO+33KalOA+aFYGlK+EfxcI5MbO7EP2yYygwh9h+s=
 gorm.io/gorm v1.25.10/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8=
+gorm.io/sharding v0.6.2 h1:V9inmbdhN+RfWPEKTvbKKKv7qxLz1CneBDQvuL5P7jg=
+gorm.io/sharding v0.6.2/go.mod h1:dXaAZv0qyUmLkLAciQ+NH2O1D1A4/ttrrZ/XK4xW9HU=
 modernc.org/libc v1.22.5 h1:91BNch/e5B0uPbJFgqbxXuOnxBQjlS//icfQEGmvyjE=
 modernc.org/libc v1.22.5/go.mod h1:jj+Z7dTNX8fBScMVNRAYZ/jF91K8fdT2hYMThc3YjBY=
 modernc.org/mathutil v1.5.0 h1:rV0Ko/6SfM+8G+yKiyI830l3Wuz1zRutdslNoQ0kfiQ=
@@ -1,6 +1,7 @@
 package main

 import (
+	"context"
 	"embed"
 	"fmt"
 	"github.com/gin-contrib/sessions"
@@ -13,6 +14,7 @@ import (
 	"openflare/middleware"
 	"openflare/model"
 	"openflare/router"
+	"openflare/service"
 	"openflare/utils/geoip"
 	"os"
 	"strconv"
@@ -67,6 +69,9 @@ func main() {
 	// Initialize options
 	model.InitOptionMap()
 	geoip.InitGeoIP()
+	backgroundCtx, cancelBackgroundTasks := context.WithCancel(context.Background())
+	defer cancelBackgroundTasks()
+	service.StartDatabaseAutoCleanupScheduler(backgroundCtx)

 	// Initialize HTTP server
 	server := gin.Default()
@@ -0,0 +1,19 @@
+package model
+
+import "time"
+
+const (
+	legacyDatabaseSchemaVersion  = 1
+	currentDatabaseSchemaVersion = 3
+	databaseSchemaVersionRowID   = 1
+)
+
+type DatabaseSchemaVersion struct {
+	ID        uint      `json:"id" gorm:"primaryKey"`
+	Version   int       `json:"version" gorm:"not null"`
+	UpdatedAt time.Time `json:"updated_at"`
+}
+
+func (DatabaseSchemaVersion) TableName() string {
+	return "database_schema_versions"
+}
@@ -1,6 +1,8 @@
 package model

 import (
+	"encoding/json"
+	"errors"
 	"fmt"
 	"github.com/glebarez/sqlite"
 	"gorm.io/driver/postgres"
@@ -22,6 +24,13 @@ 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{},
@@ -41,6 +50,12 @@ func registeredModels() []any {
 	}
 }

+func schemaMetadataModels() []any {
+	return []any{
+		&DatabaseSchemaVersion{},
+	}
+}
+
 func buildDBModels() ([]dbModel, error) {
 	models := registeredModels()
 	result := make([]dbModel, 0, len(models))
@@ -99,17 +114,13 @@ func CountTable(tableName string) (num int64) {

 func openDatabase() (*gorm.DB, string, error) {
 	if common.SQLDSN != "" {
-		db, err := gorm.Open(postgres.Open(common.SQLDSN), &gorm.Config{
-			PrepareStmt: true,
-		})
+		db, err := gorm.Open(postgres.Open(common.SQLDSN), &gorm.Config{})
 		if err != nil {
 			return nil, "", err
 		}
 		return db, "postgres", nil
 	}
-	db, err := gorm.Open(sqlite.Open(common.SQLitePath), &gorm.Config{
-		PrepareStmt: true,
-	})
+	db, err := gorm.Open(sqlite.Open(common.SQLitePath), &gorm.Config{})
 	if err != nil {
 		return nil, "", err
 	}
@@ -126,6 +137,15 @@ 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
@@ -153,10 +173,503 @@ func migrateTextColumns(db *gorm.DB, backend string) error {
 	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 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 databaseSchemaMigrations() []databaseSchemaMigration {
+	return []databaseSchemaMigration{
+		{
+			fromVersion: 1,
+			toVersion:   2,
+			migrate:     applyCurrentSchema,
+			validate:    validateDatabaseSchemaV2,
+		},
+		{
+			fromVersion: 2,
+			toVersion:   3,
+			migrate:     migrateObservabilityShardsToID,
+			validate:    validateDatabaseSchemaV3,
+		},
+	}
+}
+
+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 {
+	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 := migrateSQLiteDataIfNeeded(db, backend); err != nil {
+		return err
+	}
+	if err := validateDatabaseSchemaV3(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) {
-	for _, item := range registeredModels() {
+	models, err := buildDBModels()
+	if err != nil {
+		return false, err
+	}
+	for _, item := range models {
+		if isShardedObservabilityTable(item.tableName) {
+			for _, table := range observabilityShardTables(item.tableName) {
+				if !db.Migrator().HasTable(table) {
+					continue
+				}
+				var count int64
+				if err := db.Table(table).Limit(1).Count(&count).Error; err != nil {
+					return false, err
+				}
+				if count > 0 {
+					return false, nil
+				}
+			}
+			continue
+		}
+		if !db.Migrator().HasTable(item.value) {
+			continue
+		}
 		var count int64
-		if err := db.Model(item).Limit(1).Count(&count).Error; err != nil {
+		if err := db.Model(item.value).Limit(1).Count(&count).Error; err != nil {
 			return false, err
 		}
 		if count > 0 {
@@ -265,8 +778,17 @@ func migrateTableData(source *gorm.DB, target *gorm.DB, item dbModel) error {
 		if batchLen == 0 {
 			break
 		}
-		if err := target.Create(batchPtr.Interface()).Error; err != nil {
-			return fmt.Errorf("write target table %s failed: %w", item.tableName, err)
+		if isShardedObservabilityTable(item.tableName) {
+			for index := 0; index < batchLen; index++ {
+				record := batchPtr.Elem().Index(index)
+				if err := target.Create(record.Addr().Interface()).Error; err != nil {
+					return fmt.Errorf("write target sharded table %s failed: %w", item.tableName, err)
+				}
+			}
+		} else {
+			if err := target.Create(batchPtr.Interface()).Error; err != nil {
+				return fmt.Errorf("write target table %s failed: %w", item.tableName, err)
+			}
 		}
 		migrated += int64(batchLen)
 		offset += batchLen
@@ -293,16 +815,10 @@ func InitDB() (err error) {
 		os.Exit(1)
 	}
 	DB = db
-	if err = migrateProxyRouteEnableHTTPSColumn(db); err != nil {
+	if err = registerSharding(db, backend); err != nil {
 		return err
 	}
-	if err = autoMigrateAll(db); err != nil {
-		return err
-	}
-	if err = migrateTextColumns(db, backend); err != nil {
-		return err
-	}
-	if err = migrateSQLiteDataIfNeeded(db, backend); err != nil {
+	if err = ensureDatabaseSchemaUpToDate(db, backend); err != nil {
 		return err
 	}
 	return createRootAccountIfNeed()
@@ -1,23 +1,22 @@
 package model

 import (
+	"encoding/json"
 	"path/filepath"
 	"testing"
+	"time"

 	"github.com/glebarez/sqlite"
 	"gorm.io/gorm"
 )

-func openTestSQLiteDB(t *testing.T, name string) *gorm.DB {
+func openBareTestSQLiteDB(t *testing.T, name string) *gorm.DB {
 	t.Helper()

 	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), name)), &gorm.Config{})
 	if err != nil {
 		t.Fatalf("open sqlite db: %v", err)
 	}
-	if err := autoMigrateAll(db); err != nil {
-		t.Fatalf("auto migrate db: %v", err)
-	}
 	sqlDB, err := db.DB()
 	if err != nil {
 		t.Fatalf("get sql db: %v", err)
@@ -28,6 +27,16 @@ func openTestSQLiteDB(t *testing.T, name string) *gorm.DB {
 	return db
 }

+func openTestSQLiteDB(t *testing.T, name string) *gorm.DB {
+	t.Helper()
+
+	db := openBareTestSQLiteDB(t, name)
+	if err := autoMigrateAll(db); err != nil {
+		t.Fatalf("auto migrate db: %v", err)
+	}
+	return db
+}
+
 func findDBModelByTableName(t *testing.T, tableName string) dbModel {
 	t.Helper()

@@ -121,3 +130,307 @@ func TestMigrateTableDataCopiesRows(t *testing.T) {
 		t.Fatalf("unexpected migrated option value: %s", gotOption.Value)
 	}
 }
+
+func TestRegisterShardingAutoMigratesShardTables(t *testing.T) {
+	db := openBareTestSQLiteDB(t, "sharded.db")
+	if err := registerSharding(db, "sqlite"); err != nil {
+		t.Fatalf("register sharding: %v", err)
+	}
+	if err := autoMigrateAll(db); err != nil {
+		t.Fatalf("auto migrate db: %v", err)
+	}
+
+	for _, table := range []string{
+		"node_metric_snapshots_00",
+		"node_metric_snapshots_09",
+		"node_request_reports_00",
+		"node_request_reports_09",
+		"node_access_logs_00",
+		"node_access_logs_09",
+	} {
+		if !db.Migrator().HasTable(table) {
+			t.Fatalf("expected sharded table %s to exist", table)
+		}
+	}
+}
+
+func TestMigrateObservabilityLegacyColumnsBackfillsHealthEventMetadata(t *testing.T) {
+	db := openTestSQLiteDB(t, "legacy-health-events.db")
+
+	if err := db.Exec("ALTER TABLE node_health_events ADD COLUMN raw_json TEXT").Error; err != nil {
+		t.Fatalf("add raw_json column: %v", err)
+	}
+	rawJSON, err := json.Marshal(map[string]any{
+		"event_type": "sync_error",
+		"metadata": map[string]string{
+			"reason": "checksum_mismatch",
+			"scope":  "routes",
+		},
+	})
+	if err != nil {
+		t.Fatalf("marshal raw json: %v", err)
+	}
+	event := &NodeHealthEvent{
+		NodeID:           "node-legacy",
+		EventType:        "sync_error",
+		Severity:         "warning",
+		Status:           "active",
+		Message:          "checksum mismatch",
+		FirstTriggeredAt: time.Now().Add(-time.Minute),
+		LastTriggeredAt:  time.Now(),
+		ReportedAt:       time.Now(),
+	}
+	if err := db.Create(event).Error; err != nil {
+		t.Fatalf("create health event: %v", err)
+	}
+	if err := db.Exec("UPDATE node_health_events SET raw_json = ? WHERE id = ?", string(rawJSON), event.ID).Error; err != nil {
+		t.Fatalf("seed legacy raw_json: %v", err)
+	}
+
+	if err := migrateObservabilityLegacyColumns(db); err != nil {
+		t.Fatalf("migrateObservabilityLegacyColumns: %v", err)
+	}
+
+	var got NodeHealthEvent
+	if err := db.First(&got, event.ID).Error; err != nil {
+		t.Fatalf("query health event: %v", err)
+	}
+	if got.MetadataJSON == "" {
+		t.Fatal("expected metadata_json to be backfilled")
+	}
+}
+
+func TestEnsureDatabaseSchemaUpToDateInitializesFreshDatabase(t *testing.T) {
+	db := openBareTestSQLiteDB(t, "fresh-schema.db")
+	if err := registerSharding(db, "sqlite"); err != nil {
+		t.Fatalf("register sharding: %v", err)
+	}
+
+	if err := ensureDatabaseSchemaUpToDate(db, "sqlite"); err != nil {
+		t.Fatalf("ensureDatabaseSchemaUpToDate: %v", err)
+	}
+
+	version, exists, err := loadDatabaseSchemaVersion(db)
+	if err != nil {
+		t.Fatalf("loadDatabaseSchemaVersion: %v", err)
+	}
+	if !exists {
+		t.Fatal("expected database schema version to be recorded")
+	}
+	if version != currentDatabaseSchemaVersion {
+		t.Fatalf("unexpected schema version: got %d want %d", version, currentDatabaseSchemaVersion)
+	}
+}
+
+func TestEnsureDatabaseSchemaUpToDateUpgradesLegacyDatabase(t *testing.T) {
+	db := openBareTestSQLiteDB(t, "legacy-schema.db")
+	if err := registerSharding(db, "sqlite"); err != nil {
+		t.Fatalf("register sharding: %v", err)
+	}
+	if err := autoMigrateAll(db); err != nil {
+		t.Fatalf("auto migrate db: %v", err)
+	}
+	if err := db.Create(&User{
+		Username:    "legacy",
+		Password:    "secret",
+		DisplayName: "Legacy User",
+		Role:        1,
+		Status:      1,
+	}).Error; err != nil {
+		t.Fatalf("seed legacy user: %v", err)
+	}
+
+	if err := ensureDatabaseSchemaUpToDate(db, "sqlite"); err != nil {
+		t.Fatalf("ensureDatabaseSchemaUpToDate: %v", err)
+	}
+
+	version, exists, err := loadDatabaseSchemaVersion(db)
+	if err != nil {
+		t.Fatalf("loadDatabaseSchemaVersion: %v", err)
+	}
+	if !exists {
+		t.Fatal("expected legacy database to gain a schema version record")
+	}
+	if version != currentDatabaseSchemaVersion {
+		t.Fatalf("unexpected schema version: got %d want %d", version, currentDatabaseSchemaVersion)
+	}
+}
+
+func TestEnsureDatabaseSchemaUpToDateMigratesObservabilityShardsToID(t *testing.T) {
+	db := openBareTestSQLiteDB(t, "legacy-observability-shards.db")
+	if err := registerSharding(db, "sqlite"); err != nil {
+		t.Fatalf("register sharding: %v", err)
+	}
+	if err := autoMigrateAll(db); err != nil {
+		t.Fatalf("auto migrate db: %v", err)
+	}
+	if err := autoMigrateSchemaMetadata(db); err != nil {
+		t.Fatalf("auto migrate schema metadata: %v", err)
+	}
+
+	now := time.Now().UTC()
+	if err := db.Table("node_metric_snapshots_00").Create(&NodeMetricSnapshot{
+		ID:               1,
+		NodeID:           "node-a",
+		CapturedAt:       now.Add(-2 * time.Minute),
+		CPUUsagePercent:  22,
+		MemoryUsedBytes:  2,
+		MemoryTotalBytes: 8,
+	}).Error; err != nil {
+		t.Fatalf("seed metric snapshot shard 00: %v", err)
+	}
+	if err := db.Table("node_metric_snapshots_01").Create(&NodeMetricSnapshot{
+		ID:               1,
+		NodeID:           "node-b",
+		CapturedAt:       now.Add(-time.Minute),
+		CPUUsagePercent:  44,
+		MemoryUsedBytes:  4,
+		MemoryTotalBytes: 8,
+	}).Error; err != nil {
+		t.Fatalf("seed metric snapshot shard 01: %v", err)
+	}
+	if err := db.Table("node_request_reports_00").Create(&NodeRequestReport{
+		ID:                 1,
+		NodeID:             "node-a",
+		WindowStartedAt:    now.Add(-3 * time.Minute),
+		WindowEndedAt:      now.Add(-2 * time.Minute),
+		RequestCount:       12,
+		ErrorCount:         1,
+		UniqueVisitorCount: 6,
+	}).Error; err != nil {
+		t.Fatalf("seed request report shard 00: %v", err)
+	}
+	if err := db.Table("node_request_reports_01").Create(&NodeRequestReport{
+		ID:                 1,
+		NodeID:             "node-b",
+		WindowStartedAt:    now.Add(-2 * time.Minute),
+		WindowEndedAt:      now.Add(-time.Minute),
+		RequestCount:       21,
+		ErrorCount:         2,
+		UniqueVisitorCount: 9,
+	}).Error; err != nil {
+		t.Fatalf("seed request report shard 01: %v", err)
+	}
+	if err := db.Table("node_access_logs_00").Create(&NodeAccessLog{
+		ID:         1,
+		NodeID:     "node-a",
+		LoggedAt:   now.Add(-90 * time.Second),
+		RemoteAddr: "203.0.113.10",
+		Host:       "a.example.com",
+		Path:       "/alpha",
+		StatusCode: 200,
+	}).Error; err != nil {
+		t.Fatalf("seed access log shard 00: %v", err)
+	}
+	if err := db.Table("node_access_logs_01").Create(&NodeAccessLog{
+		ID:         1,
+		NodeID:     "node-b",
+		LoggedAt:   now.Add(-60 * time.Second),
+		RemoteAddr: "203.0.113.11",
+		Host:       "b.example.com",
+		Path:       "/beta",
+		StatusCode: 502,
+	}).Error; err != nil {
+		t.Fatalf("seed access log shard 01: %v", err)
+	}
+	if err := saveDatabaseSchemaVersion(db, 2); err != nil {
+		t.Fatalf("save schema version: %v", err)
+	}
+
+	previousDB := DB
+	DB = db
+	t.Cleanup(func() {
+		DB = previousDB
+	})
+
+	if err := ensureDatabaseSchemaUpToDate(db, "sqlite"); err != nil {
+		t.Fatalf("ensureDatabaseSchemaUpToDate: %v", err)
+	}
+
+	version, exists, err := loadDatabaseSchemaVersion(db)
+	if err != nil {
+		t.Fatalf("loadDatabaseSchemaVersion: %v", err)
+	}
+	if !exists {
+		t.Fatal("expected migrated database to keep schema version record")
+	}
+	if version != currentDatabaseSchemaVersion {
+		t.Fatalf("unexpected schema version: got %d want %d", version, currentDatabaseSchemaVersion)
+	}
+
+	for _, baseTable := range shardedObservabilityBaseTables() {
+		for _, table := range observabilityShardTables(baseTable) {
+			legacyTable := legacyObservabilityShardTableName(table)
+			if db.Migrator().HasTable(legacyTable) {
+				t.Fatalf("expected legacy shard table %s to be removed", legacyTable)
+			}
+		}
+	}
+
+	snapshots, err := ListMetricSnapshotsSince(time.Time{})
+	if err != nil {
+		t.Fatalf("ListMetricSnapshotsSince failed: %v", err)
+	}
+	if len(snapshots) != 2 {
+		t.Fatalf("expected 2 migrated metric snapshots, got %+v", snapshots)
+	}
+	reports, err := ListRequestReportsSince(time.Time{})
+	if err != nil {
+		t.Fatalf("ListRequestReportsSince failed: %v", err)
+	}
+	if len(reports) != 2 {
+		t.Fatalf("expected 2 migrated request reports, got %+v", reports)
+	}
+	logs, err := ListNodeAccessLogs(NodeAccessLogQuery{Page: 0, PageSize: 10})
+	if err != nil {
+		t.Fatalf("ListNodeAccessLogs failed: %v", err)
+	}
+	if len(logs) != 2 {
+		t.Fatalf("expected 2 migrated access logs, got %+v", logs)
+	}
+
+	seenSnapshotIDs := make(map[uint]struct{}, len(snapshots))
+	for _, item := range snapshots {
+		if item == nil || item.ID == 0 {
+			t.Fatalf("expected migrated metric snapshot to have a new non-zero id: %+v", item)
+		}
+		if _, exists := seenSnapshotIDs[item.ID]; exists {
+			t.Fatalf("expected migrated metric snapshot ids to be unique, got duplicate %d", item.ID)
+		}
+		seenSnapshotIDs[item.ID] = struct{}{}
+		targetTable := observabilityShardTableForID("node_metric_snapshots", item.ID)
+		var count int64
+		if err := db.Table(targetTable).Where("id = ?", item.ID).Count(&count).Error; err != nil {
+			t.Fatalf("count migrated metric snapshot in target shard: %v", err)
+		}
+		if count != 1 {
+			t.Fatalf("expected migrated metric snapshot id %d to be stored in %s", item.ID, targetTable)
+		}
+	}
+}
+
+func TestRunDatabaseSchemaMigrationDoesNotAdvanceVersionWhenValidationFails(t *testing.T) {
+	db := openBareTestSQLiteDB(t, "failed-validation.db")
+
+	err := runDatabaseSchemaMigration(db, "sqlite", databaseSchemaMigration{
+		fromVersion: legacyDatabaseSchemaVersion,
+		toVersion:   currentDatabaseSchemaVersion,
+		migrate: func(tx *gorm.DB, backend string) error {
+			return autoMigrateSchemaMetadata(tx)
+		},
+		validate: func(tx *gorm.DB, backend string) error {
+			return gorm.ErrInvalidDB
+		},
+	})
+	if err == nil {
+		t.Fatal("expected migration validation to fail")
+	}
+
+	_, exists, loadErr := loadDatabaseSchemaVersion(db)
+	if loadErr != nil {
+		t.Fatalf("loadDatabaseSchemaVersion: %v", loadErr)
+	}
+	if exists {
+		t.Fatal("expected schema version to remain unset after failed validation")
+	}
+}
@@ -2,6 +2,7 @@ package model

 import (
 	"fmt"
+	"sort"
 	"strings"
 	"time"

@@ -10,14 +11,13 @@ import (

 type NodeAccessLog struct {
 	ID         uint      `json:"id" gorm:"primaryKey"`
-	NodeID     string    `json:"node_id" gorm:"index:idx_node_access_logs_node_logged_at,priority:1;size:64;not null"`
-	LoggedAt   time.Time `json:"logged_at" gorm:"index:idx_node_access_logs_logged_at;index:idx_node_access_logs_node_logged_at,priority:2"`
-	RemoteAddr string    `json:"remote_addr" gorm:"index:idx_node_access_logs_remote_addr;size:128"`
+	NodeID     string    `json:"node_id" gorm:"index:,composite:node_logged_at,priority:1;size:64;not null"`
+	LoggedAt   time.Time `json:"logged_at" gorm:"index;index:,composite:node_logged_at,priority:2"`
+	RemoteAddr string    `json:"remote_addr" gorm:"index;size:128"`
 	Region     string    `json:"region" gorm:"size:128"`
-	Host       string    `json:"host" gorm:"index:idx_node_access_logs_host;size:255"`
+	Host       string    `json:"host" gorm:"index;size:255"`
 	Path       string    `json:"path" gorm:"size:2048"`
-	StatusCode int       `json:"status_code" gorm:"index:idx_node_access_logs_status_code"`
-	RawJSON    string    `json:"raw_json" gorm:"type:text"`
+	StatusCode int       `json:"status_code" gorm:"index"`
 	CreatedAt  time.Time `json:"created_at"`
 }

@@ -92,85 +92,92 @@ type NodeAccessLogTrendPointRow struct {
 	RequestCount int64 `json:"request_count"`
 }

+func (log *NodeAccessLog) BeforeCreate(tx *gorm.DB) error {
+	return assignObservabilityID(&log.ID)
+}
+
 func ListNodeAccessLogs(query NodeAccessLogQuery) (logs []*NodeAccessLog, err error) {
-	offset := query.Page * query.PageSize
-	db := buildNodeAccessLogQuery(DB, query).
-		Order(buildNodeAccessLogSortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset)
-	err = db.Find(&logs).Error
-	return logs, err
+	all, err := listNodeAccessLogsAcrossShards(query)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(all), query.Page, query.PageSize)
+	if start >= len(all) {
+		return []*NodeAccessLog{}, nil
+	}
+	return all[start:end], nil
 }

 func CountNodeAccessLogs(query NodeAccessLogQuery) (totalRecords int64, totalIPs int64, err error) {
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), query)
-	if err = base.Count(&totalRecords).Error; err != nil {
+	all, err := listNodeAccessLogsAcrossShards(query)
+	if err != nil {
 		return 0, 0, err
 	}
-	distinctQuery := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), query).
-		Where("remote_addr <> ''").
-		Distinct("remote_addr")
-	if err = distinctQuery.Count(&totalIPs).Error; err != nil {
-		return 0, 0, err
+	ips := make(map[string]struct{}, len(all))
+	for _, item := range all {
+		if item == nil {
+			continue
+		}
+		trimmed := strings.TrimSpace(item.RemoteAddr)
+		if trimmed != "" {
+			ips[trimmed] = struct{}{}
+		}
 	}
-	return totalRecords, totalIPs, nil
+	return int64(len(all)), int64(len(ips)), nil
 }

 func ListNodeAccessLogRegionCounts(nodeID string, since time.Time, limit int) (items []*NodeAccessLogRegionCount, err error) {
-	query := DB.Model(&NodeAccessLog{}).
-		Select("region as region, count(*) as count").
-		Where("region <> ''")
-	if nodeID != "" {
-		query = query.Where("node_id = ?", nodeID)
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID: nodeID,
+		Since:  since,
+	})
+	if err != nil {
+		return nil, err
 	}
-	if !since.IsZero() {
-		query = query.Where("logged_at >= ?", since)
+	counts := make(map[string]int64)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		region := strings.TrimSpace(item.Region)
+		if region == "" {
+			continue
+		}
+		counts[region]++
 	}
-	query = query.Group("region").Order("count desc, region asc")
-	if limit > 0 {
-		query = query.Limit(limit)
+	items = make([]*NodeAccessLogRegionCount, 0, len(counts))
+	for region, count := range counts {
+		items = append(items, &NodeAccessLogRegionCount{
+			Region: region,
+			Count:  count,
+		})
 	}
-	err = query.Scan(&items).Error
-	return items, err
+	sort.Slice(items, func(i int, j int) bool {
+		if items[i].Count == items[j].Count {
+			return items[i].Region < items[j].Region
+		}
+		return items[i].Count > items[j].Count
+	})
+	if limit > 0 && len(items) > limit {
+		items = items[:limit]
+	}
+	return items, nil
 }

 func ListNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (items []*NodeAccessLogBucketRow, err error) {
-	offset := query.Page * query.PageSize
-	bucketExpr := accessLogBucketEpochExpr(query.FoldMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Path:       query.Path,
-		Since:      query.Since,
-	})
-	err = base.Select(fmt.Sprintf(
-		"%s as bucket_epoch, count(*) as request_count, count(distinct remote_addr) as unique_ip_count, count(distinct host) as unique_host_count, sum(case when status_code < 400 then 1 else 0 end) as success_count, sum(case when status_code >= 400 and status_code < 500 then 1 else 0 end) as client_error_count, sum(case when status_code >= 500 then 1 else 0 end) as server_error_count",
-		bucketExpr,
-	)).
-		Group(bucketExpr).
-		Order(buildNodeAccessLogBucketSortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset).
-		Scan(&items).Error
-	return items, err
+	rows, err := buildNodeAccessLogBucketRows(query)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(rows), query.Page, query.PageSize)
+	if start >= len(rows) {
+		return []*NodeAccessLogBucketRow{}, nil
+	}
+	return rows[start:end], nil
 }

 func CountNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (total int64, err error) {
-	bucketExpr := accessLogBucketEpochExpr(query.FoldMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Path:       query.Path,
-		Since:      query.Since,
-	})
-	rows := []struct {
-		BucketEpoch int64 `gorm:"column:bucket_epoch"`
-	}{}
-	err = base.Select(fmt.Sprintf("%s as bucket_epoch", bucketExpr)).
-		Group(bucketExpr).
-		Scan(&rows).Error
+	rows, err := buildNodeAccessLogBucketRows(query)
 	if err != nil {
 		return 0, err
 	}
@@ -178,39 +185,19 @@ func CountNodeAccessLogBuckets(query NodeAccessLogBucketQuery) (total int64, err
 }

 func ListNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery, recentSince time.Time) (items []*NodeAccessLogIPSummaryRow, err error) {
-	offset := query.Page * query.PageSize
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Since:      query.Since,
-	}).Where("remote_addr <> ''")
-	lastSeenExpr := accessLogEpochExpr("max(logged_at)")
-	err = base.Select(
-		"remote_addr as remote_addr, count(*) as total_requests, sum(case when logged_at >= ? then 1 else 0 end) as recent_requests, "+lastSeenExpr+" as last_seen_epoch",
-		recentSince,
-	).
-		Group("remote_addr").
-		Order(buildNodeAccessLogIPSummarySortClause(query.SortBy, query.SortOrder)).
-		Limit(query.PageSize).
-		Offset(offset).
-		Scan(&items).Error
-	return items, err
+	rows, err := buildNodeAccessLogIPSummaryRows(query, recentSince)
+	if err != nil {
+		return nil, err
+	}
+	start, end := paginateBounds(len(rows), query.Page, query.PageSize)
+	if start >= len(rows) {
+		return []*NodeAccessLogIPSummaryRow{}, nil
+	}
+	return rows[start:end], nil
 }

 func CountNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery) (total int64, err error) {
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
-		NodeID:     query.NodeID,
-		RemoteAddr: query.RemoteAddr,
-		Host:       query.Host,
-		Since:      query.Since,
-	}).Where("remote_addr <> ''")
-	rows := []struct {
-		RemoteAddr string `gorm:"column:remote_addr"`
-	}{}
-	err = base.Select("remote_addr").
-		Group("remote_addr").
-		Scan(&rows).Error
+	rows, err := buildNodeAccessLogIPSummaryRows(query, time.Time{})
 	if err != nil {
 		return 0, err
 	}
@@ -218,23 +205,82 @@ func CountNodeAccessLogIPSummaries(query NodeAccessLogIPSummaryQuery) (total int
 }

 func ListNodeAccessLogIPTrend(query NodeAccessLogIPTrendQuery) (items []*NodeAccessLogTrendPointRow, err error) {
-	bucketExpr := accessLogBucketEpochExpr(query.BucketMinutes)
-	base := buildNodeAccessLogQuery(DB.Model(&NodeAccessLog{}), NodeAccessLogQuery{
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
 		NodeID:     query.NodeID,
 		RemoteAddr: query.RemoteAddr,
 		Host:       query.Host,
 		Since:      query.Since,
-	}).Where("remote_addr = ?", strings.TrimSpace(query.RemoteAddr))
-	err = base.Select(fmt.Sprintf("%s as bucket_epoch, count(*) as request_count", bucketExpr)).
-		Group(bucketExpr).
-		Order("bucket_epoch asc").
-		Scan(&items).Error
-	return items, err
+	})
+	if err != nil {
+		return nil, err
+	}
+	remoteAddr := strings.TrimSpace(query.RemoteAddr)
+	if remoteAddr == "" {
+		return []*NodeAccessLogTrendPointRow{}, nil
+	}
+	buckets := make(map[int64]int64)
+	for _, item := range logs {
+		if item == nil || strings.TrimSpace(item.RemoteAddr) != remoteAddr {
+			continue
+		}
+		bucketEpoch := bucketEpochForTime(item.LoggedAt, query.BucketMinutes)
+		buckets[bucketEpoch]++
+	}
+	items = make([]*NodeAccessLogTrendPointRow, 0, len(buckets))
+	for bucketEpoch, requestCount := range buckets {
+		items = append(items, &NodeAccessLogTrendPointRow{
+			BucketEpoch:  bucketEpoch,
+			RequestCount: requestCount,
+		})
+	}
+	sort.Slice(items, func(i int, j int) bool {
+		return items[i].BucketEpoch < items[j].BucketEpoch
+	})
+	return items, nil
 }

 func DeleteNodeAccessLogsBefore(before time.Time) (deleted int64, err error) {
-	result := DB.Where("logged_at < ?", before).Delete(&NodeAccessLog{})
-	return result.RowsAffected, result.Error
+	return deleteAcrossShards(DB, "node_access_logs", &NodeAccessLog{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("logged_at < ?", before)
+	})
+}
+
+func DeleteAllNodeAccessLogs(db *gorm.DB) (deleted int64, err error) {
+	return deleteAcrossShards(db, "node_access_logs", &NodeAccessLog{}, nil)
+}
+
+func NodeAccessLogExists(db *gorm.DB, record *NodeAccessLog) (bool, error) {
+	if record == nil {
+		return false, nil
+	}
+	db = normalizeShardedDB(db)
+	for _, table := range observabilityShardTables("node_access_logs") {
+		var count int64
+		if err := db.Table(table).
+			Where(
+				"node_id = ? AND logged_at = ? AND remote_addr = ? AND host = ? AND path = ? AND status_code = ?",
+				record.NodeID,
+				record.LoggedAt,
+				record.RemoteAddr,
+				record.Host,
+				record.Path,
+				record.StatusCode,
+			).
+			Limit(1).
+			Count(&count).Error; err != nil {
+			return false, err
+		}
+		if count > 0 {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func DeleteNodeAccessLogsByNodeBefore(db *gorm.DB, nodeID string, before time.Time) (deleted int64, err error) {
+	return deleteAcrossShards(db, "node_access_logs", &NodeAccessLog{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("node_id = ? AND logged_at < ?", nodeID, before)
+	})
 }

 func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
@@ -244,6 +290,10 @@ func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	if db.Statement == nil || db.Statement.Model == nil {
 		db = db.Model(&NodeAccessLog{})
 	}
+	return applyNodeAccessLogFilters(db, query)
+}
+
+func applyNodeAccessLogFilters(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	if trimmed := strings.TrimSpace(query.NodeID); trimmed != "" {
 		db = db.Where("node_id LIKE ?", "%"+trimmed+"%")
 	}
@@ -262,6 +312,297 @@ func buildNodeAccessLogQuery(db *gorm.DB, query NodeAccessLogQuery) *gorm.DB {
 	return db
 }

+func listNodeAccessLogsAcrossShards(query NodeAccessLogQuery) ([]*NodeAccessLog, error) {
+	items, err := queryAcrossShards("node_access_logs", func(tx *gorm.DB) ([]*NodeAccessLog, error) {
+		var shardRows []*NodeAccessLog
+		if err := applyNodeAccessLogFilters(tx, query).Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	sortNodeAccessLogs(items, query.SortBy, query.SortOrder)
+	return items, nil
+}
+
+func buildNodeAccessLogBucketRows(query NodeAccessLogBucketQuery) ([]*NodeAccessLogBucketRow, error) {
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID:     query.NodeID,
+		RemoteAddr: query.RemoteAddr,
+		Host:       query.Host,
+		Path:       query.Path,
+		Since:      query.Since,
+	})
+	if err != nil {
+		return nil, err
+	}
+	type bucketAccumulator struct {
+		requestCount     int64
+		uniqueIPs        map[string]struct{}
+		uniqueHosts      map[string]struct{}
+		successCount     int64
+		clientErrorCount int64
+		serverErrorCount int64
+	}
+	accumulators := make(map[int64]*bucketAccumulator)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		bucketEpoch := bucketEpochForTime(item.LoggedAt, query.FoldMinutes)
+		accumulator := accumulators[bucketEpoch]
+		if accumulator == nil {
+			accumulator = &bucketAccumulator{
+				uniqueIPs:   make(map[string]struct{}),
+				uniqueHosts: make(map[string]struct{}),
+			}
+			accumulators[bucketEpoch] = accumulator
+		}
+		accumulator.requestCount++
+		if trimmed := strings.TrimSpace(item.RemoteAddr); trimmed != "" {
+			accumulator.uniqueIPs[trimmed] = struct{}{}
+		}
+		if trimmed := strings.TrimSpace(item.Host); trimmed != "" {
+			accumulator.uniqueHosts[trimmed] = struct{}{}
+		}
+		switch {
+		case item.StatusCode < 400:
+			accumulator.successCount++
+		case item.StatusCode < 500:
+			accumulator.clientErrorCount++
+		default:
+			accumulator.serverErrorCount++
+		}
+	}
+	rows := make([]*NodeAccessLogBucketRow, 0, len(accumulators))
+	for bucketEpoch, accumulator := range accumulators {
+		rows = append(rows, &NodeAccessLogBucketRow{
+			BucketEpoch:      bucketEpoch,
+			RequestCount:     accumulator.requestCount,
+			UniqueIPCount:    int64(len(accumulator.uniqueIPs)),
+			UniqueHostCount:  int64(len(accumulator.uniqueHosts)),
+			SuccessCount:     accumulator.successCount,
+			ClientErrorCount: accumulator.clientErrorCount,
+			ServerErrorCount: accumulator.serverErrorCount,
+		})
+	}
+	sortNodeAccessLogBucketRows(rows, query.SortBy, query.SortOrder)
+	return rows, nil
+}
+
+func buildNodeAccessLogIPSummaryRows(query NodeAccessLogIPSummaryQuery, recentSince time.Time) ([]*NodeAccessLogIPSummaryRow, error) {
+	logs, err := listNodeAccessLogsAcrossShards(NodeAccessLogQuery{
+		NodeID:     query.NodeID,
+		RemoteAddr: query.RemoteAddr,
+		Host:       query.Host,
+		Since:      query.Since,
+	})
+	if err != nil {
+		return nil, err
+	}
+	type accumulator struct {
+		totalRequests  int64
+		recentRequests int64
+		lastSeenAt     time.Time
+	}
+	accumulators := make(map[string]*accumulator)
+	for _, item := range logs {
+		if item == nil {
+			continue
+		}
+		remoteAddr := strings.TrimSpace(item.RemoteAddr)
+		if remoteAddr == "" {
+			continue
+		}
+		acc := accumulators[remoteAddr]
+		if acc == nil {
+			acc = &accumulator{}
+			accumulators[remoteAddr] = acc
+		}
+		acc.totalRequests++
+		if !recentSince.IsZero() && !item.LoggedAt.Before(recentSince) {
+			acc.recentRequests++
+		}
+		if item.LoggedAt.After(acc.lastSeenAt) {
+			acc.lastSeenAt = item.LoggedAt
+		}
+	}
+	rows := make([]*NodeAccessLogIPSummaryRow, 0, len(accumulators))
+	for remoteAddr, acc := range accumulators {
+		rows = append(rows, &NodeAccessLogIPSummaryRow{
+			RemoteAddr:     remoteAddr,
+			TotalRequests:  acc.totalRequests,
+			RecentRequests: acc.recentRequests,
+			LastSeenEpoch:  acc.lastSeenAt.Unix(),
+		})
+	}
+	sortNodeAccessLogIPSummaryRows(rows, query.SortBy, query.SortOrder)
+	return rows, nil
+}
+
+func sortNodeAccessLogs(items []*NodeAccessLog, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "status_code":
+			compare = compareInt(left.StatusCode, right.StatusCode)
+		case "remote_addr":
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		case "host":
+			compare = strings.Compare(left.Host, right.Host)
+		case "path":
+			compare = strings.Compare(left.Path, right.Path)
+		default:
+			compare = compareTime(left.LoggedAt, right.LoggedAt)
+		}
+		if compare == 0 {
+			compare = compareTime(left.LoggedAt, right.LoggedAt)
+		}
+		if compare == 0 {
+			compare = compareUint(left.ID, right.ID)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func sortNodeAccessLogBucketRows(items []*NodeAccessLogBucketRow, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "request_count":
+			compare = compareInt64(left.RequestCount, right.RequestCount)
+		default:
+			compare = compareInt64(left.BucketEpoch, right.BucketEpoch)
+		}
+		if compare == 0 {
+			compare = compareInt64(left.BucketEpoch, right.BucketEpoch)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func sortNodeAccessLogIPSummaryRows(items []*NodeAccessLogIPSummaryRow, sortBy string, sortOrder string) {
+	desc := normalizeSortOrder(sortOrder) != "asc"
+	sort.Slice(items, func(i int, j int) bool {
+		left := items[i]
+		right := items[j]
+		if left == nil || right == nil {
+			return left != nil
+		}
+		var compare int
+		switch strings.TrimSpace(sortBy) {
+		case "recent_requests":
+			compare = compareInt64(left.RecentRequests, right.RecentRequests)
+		case "last_seen_at":
+			compare = compareInt64(left.LastSeenEpoch, right.LastSeenEpoch)
+		case "remote_addr":
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		default:
+			compare = compareInt64(left.TotalRequests, right.TotalRequests)
+		}
+		if compare == 0 {
+			compare = compareInt64(left.LastSeenEpoch, right.LastSeenEpoch)
+		}
+		if compare == 0 {
+			compare = strings.Compare(left.RemoteAddr, right.RemoteAddr)
+		}
+		if desc {
+			return compare > 0
+		}
+		return compare < 0
+	})
+}
+
+func paginateBounds(total int, page int, pageSize int) (int, int) {
+	if page < 0 {
+		page = 0
+	}
+	if pageSize <= 0 {
+		return 0, total
+	}
+	start := page * pageSize
+	if start > total {
+		start = total
+	}
+	end := start + pageSize
+	if end > total {
+		end = total
+	}
+	return start, end
+}
+
+func bucketEpochForTime(value time.Time, bucketMinutes int) int64 {
+	bucketSeconds := int64(bucketMinutes * 60)
+	if bucketSeconds <= 0 {
+		bucketSeconds = 180
+	}
+	return (value.UTC().Unix() / bucketSeconds) * bucketSeconds
+}
+
+func compareTime(left time.Time, right time.Time) int {
+	switch {
+	case left.After(right):
+		return 1
+	case left.Before(right):
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareInt(left int, right int) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareInt64(left int64, right int64) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
+func compareUint(left uint, right uint) int {
+	switch {
+	case left > right:
+		return 1
+	case left < right:
+		return -1
+	default:
+		return 0
+	}
+}
+
 func buildNodeAccessLogSortClause(sortBy string, sortOrder string) string {
 	column := "logged_at"
 	switch strings.TrimSpace(sortBy) {
@@ -13,7 +13,7 @@ type NodeHealthEvent struct {
 	LastTriggeredAt  time.Time  `json:"last_triggered_at" gorm:"index"`
 	ReportedAt       time.Time  `json:"reported_at" gorm:"index"`
 	ResolvedAt       *time.Time `json:"resolved_at" gorm:"index"`
-	RawJSON          string     `json:"raw_json" gorm:"type:text"`
+	MetadataJSON     string     `json:"metadata_json" gorm:"type:text"`
 	CreatedAt        time.Time  `json:"created_at"`
 	UpdatedAt        time.Time  `json:"updated_at"`
 }
@@ -1,6 +1,11 @@
 package model

-import "time"
+import (
+	"sort"
+	"time"
+
+	"gorm.io/gorm"
+)

 type NodeMetricSnapshot struct {
 	ID                   uint      `json:"id" gorm:"primaryKey"`
@@ -18,31 +23,94 @@ type NodeMetricSnapshot struct {
 	OpenrestyRxBytes     int64     `json:"openresty_rx_bytes"`
 	OpenrestyTxBytes     int64     `json:"openresty_tx_bytes"`
 	OpenrestyConnections int64     `json:"openresty_connections"`
-	RawJSON              string    `json:"raw_json" gorm:"type:text"`
 	CreatedAt            time.Time `json:"created_at"`
 }

+func (snapshot *NodeMetricSnapshot) BeforeCreate(tx *gorm.DB) error {
+	return assignObservabilityID(&snapshot.ID)
+}
+
 func (snapshot *NodeMetricSnapshot) Insert() error {
 	return DB.Create(snapshot).Error
 }

 func ListNodeMetricSnapshots(nodeID string, since time.Time, limit int) (snapshots []*NodeMetricSnapshot, err error) {
-	query := DB.Where("node_id = ?", nodeID).Order("captured_at desc")
-	if !since.IsZero() {
-		query = query.Where("captured_at >= ?", since)
+	rows, err := queryAcrossShards("node_metric_snapshots", func(tx *gorm.DB) ([]*NodeMetricSnapshot, error) {
+		var shardRows []*NodeMetricSnapshot
+		query := tx.Order("captured_at desc, id desc")
+		if nodeID != "" {
+			query = query.Where("node_id = ?", nodeID)
+		}
+		if !since.IsZero() {
+			query = query.Where("captured_at >= ?", since)
+		}
+		if err := query.Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-	if limit > 0 {
-		query = query.Limit(limit)
+	sort.Slice(rows, func(i int, j int) bool {
+		if rows[i].CapturedAt.Equal(rows[j].CapturedAt) {
+			return rows[i].ID > rows[j].ID
+		}
+		return rows[i].CapturedAt.After(rows[j].CapturedAt)
+	})
+	if limit > 0 && len(rows) > limit {
+		rows = rows[:limit]
 	}
-	err = query.Find(&snapshots).Error
-	return snapshots, err
+	return rows, nil
 }

 func ListMetricSnapshotsSince(since time.Time) (snapshots []*NodeMetricSnapshot, err error) {
-	query := DB.Order("captured_at desc")
-	if !since.IsZero() {
-		query = query.Where("captured_at >= ?", since)
+	rows, err := queryAcrossShards("node_metric_snapshots", func(tx *gorm.DB) ([]*NodeMetricSnapshot, error) {
+		var shardRows []*NodeMetricSnapshot
+		query := tx.Order("captured_at desc")
+		if !since.IsZero() {
+			query = query.Where("captured_at >= ?", since)
+		}
+		if err := query.Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-	err = query.Find(&snapshots).Error
-	return snapshots, err
+	sort.Slice(rows, func(i int, j int) bool {
+		if rows[i].CapturedAt.Equal(rows[j].CapturedAt) {
+			return rows[i].ID > rows[j].ID
+		}
+		return rows[i].CapturedAt.After(rows[j].CapturedAt)
+	})
+	return rows, nil
+}
+
+func NodeMetricSnapshotExists(db *gorm.DB, nodeID string, capturedAt time.Time) (bool, error) {
+	db = normalizeShardedDB(db)
+	for _, table := range observabilityShardTables("node_metric_snapshots") {
+		var count int64
+		if err := db.Table(table).
+			Where("node_id = ? AND captured_at = ?", nodeID, capturedAt).
+			Limit(1).
+			Count(&count).Error; err != nil {
+			return false, err
+		}
+		if count > 0 {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func DeleteNodeMetricSnapshotsBefore(db *gorm.DB, before time.Time) (int64, error) {
+	return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("captured_at < ?", before)
+	})
+}
+
+func DeleteAllNodeMetricSnapshots(db *gorm.DB) (int64, error) {
+	return deleteAcrossShards(db, "node_metric_snapshots", &NodeMetricSnapshot{}, nil)
 }
@@ -1,6 +1,11 @@
 package model

-import "time"
+import (
+	"sort"
+	"time"
+
+	"gorm.io/gorm"
+)

 type NodeRequestReport struct {
 	ID                  uint      `json:"id" gorm:"primaryKey"`
@@ -13,31 +18,94 @@ type NodeRequestReport struct {
 	StatusCodesJSON     string    `json:"status_codes_json" gorm:"type:text"`
 	TopDomainsJSON      string    `json:"top_domains_json" gorm:"type:text"`
 	SourceCountriesJSON string    `json:"source_countries_json" gorm:"type:text"`
-	RawJSON             string    `json:"raw_json" gorm:"type:text"`
 	CreatedAt           time.Time `json:"created_at"`
 }

+func (report *NodeRequestReport) BeforeCreate(tx *gorm.DB) error {
+	return assignObservabilityID(&report.ID)
+}
+
 func (report *NodeRequestReport) Insert() error {
 	return DB.Create(report).Error
 }

 func ListNodeRequestReports(nodeID string, since time.Time, limit int) (reports []*NodeRequestReport, err error) {
-	query := DB.Where("node_id = ?", nodeID).Order("window_ended_at desc")
-	if !since.IsZero() {
-		query = query.Where("window_ended_at >= ?", since)
+	rows, err := queryAcrossShards("node_request_reports", func(tx *gorm.DB) ([]*NodeRequestReport, error) {
+		var shardRows []*NodeRequestReport
+		query := tx.Order("window_ended_at desc, id desc")
+		if nodeID != "" {
+			query = query.Where("node_id = ?", nodeID)
+		}
+		if !since.IsZero() {
+			query = query.Where("window_ended_at >= ?", since)
+		}
+		if err := query.Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-	if limit > 0 {
-		query = query.Limit(limit)
+	sort.Slice(rows, func(i int, j int) bool {
+		if rows[i].WindowEndedAt.Equal(rows[j].WindowEndedAt) {
+			return rows[i].ID > rows[j].ID
+		}
+		return rows[i].WindowEndedAt.After(rows[j].WindowEndedAt)
+	})
+	if limit > 0 && len(rows) > limit {
+		rows = rows[:limit]
 	}
-	err = query.Find(&reports).Error
-	return reports, err
+	return rows, nil
 }

 func ListRequestReportsSince(since time.Time) (reports []*NodeRequestReport, err error) {
-	query := DB.Order("window_ended_at desc")
-	if !since.IsZero() {
-		query = query.Where("window_ended_at >= ?", since)
+	rows, err := queryAcrossShards("node_request_reports", func(tx *gorm.DB) ([]*NodeRequestReport, error) {
+		var shardRows []*NodeRequestReport
+		query := tx.Order("window_ended_at desc")
+		if !since.IsZero() {
+			query = query.Where("window_ended_at >= ?", since)
+		}
+		if err := query.Find(&shardRows).Error; err != nil {
+			return nil, err
+		}
+		return shardRows, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-	err = query.Find(&reports).Error
-	return reports, err
+	sort.Slice(rows, func(i int, j int) bool {
+		if rows[i].WindowEndedAt.Equal(rows[j].WindowEndedAt) {
+			return rows[i].ID > rows[j].ID
+		}
+		return rows[i].WindowEndedAt.After(rows[j].WindowEndedAt)
+	})
+	return rows, nil
+}
+
+func NodeRequestReportExists(db *gorm.DB, nodeID string, windowStartedAt time.Time, windowEndedAt time.Time) (bool, error) {
+	db = normalizeShardedDB(db)
+	for _, table := range observabilityShardTables("node_request_reports") {
+		var count int64
+		if err := db.Table(table).
+			Where("node_id = ? AND window_started_at = ? AND window_ended_at = ?", nodeID, windowStartedAt, windowEndedAt).
+			Limit(1).
+			Count(&count).Error; err != nil {
+			return false, err
+		}
+		if count > 0 {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func DeleteNodeRequestReportsBefore(db *gorm.DB, before time.Time) (int64, error) {
+	return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, func(tx *gorm.DB) *gorm.DB {
+		return tx.Where("window_ended_at < ?", before)
+	})
+}
+
+func DeleteAllNodeRequestReports(db *gorm.DB) (int64, error) {
+	return deleteAcrossShards(db, "node_request_reports", &NodeRequestReport{}, nil)
 }
@@ -20,7 +20,6 @@ type NodeSystemProfile struct {
 	TotalDiskBytes   int64     `json:"total_disk_bytes"`
 	UptimeSeconds    int64     `json:"uptime_seconds"`
 	ReportedAt       time.Time `json:"reported_at" gorm:"index"`
-	RawJSON          string    `json:"raw_json" gorm:"type:text"`
 	CreatedAt        time.Time `json:"created_at"`
 	UpdatedAt        time.Time `json:"updated_at"`
 }
@@ -49,7 +48,6 @@ func UpsertNodeSystemProfile(profile *NodeSystemProfile) error {
 			"total_disk_bytes",
 			"uptime_seconds",
 			"reported_at",
-			"raw_json",
 			"updated_at",
 		}),
 	}).Create(profile).Error
@@ -56,6 +56,8 @@ func InitOptionMap() {
 	common.OptionMap["NodeOfflineThreshold"] = strconv.Itoa(int(common.NodeOfflineThreshold.Milliseconds()))
 	common.OptionMap["AgentUpdateRepo"] = common.AgentUpdateRepo
 	common.OptionMap["GeoIPProvider"] = common.GeoIPProvider
+	common.OptionMap["DatabaseAutoCleanupEnabled"] = strconv.FormatBool(common.DatabaseAutoCleanupEnabled)
+	common.OptionMap["DatabaseAutoCleanupRetentionDays"] = strconv.Itoa(common.DatabaseAutoCleanupRetentionDays)
 	common.OptionMap["OpenRestyWorkerProcesses"] = common.OpenRestyWorkerProcesses
 	common.OptionMap["OpenRestyWorkerConnections"] = strconv.Itoa(common.OpenRestyWorkerConnections)
 	common.OptionMap["OpenRestyWorkerRlimitNofile"] = strconv.Itoa(common.OpenRestyWorkerRlimitNofile)
@@ -219,6 +221,12 @@ func updateOptionMap(key string, value string) {
 			common.GeoIPProvider = value
 			shouldRefreshGeoIP = true
 		}
+	case "DatabaseAutoCleanupEnabled":
+		common.DatabaseAutoCleanupEnabled = value == "true"
+	case "DatabaseAutoCleanupRetentionDays":
+		if v, err := strconv.Atoi(value); err == nil && v >= 1 {
+			common.DatabaseAutoCleanupRetentionDays = v
+		}
 	case "OpenRestyWorkerProcesses":
 		if strings.TrimSpace(value) != "" {
 			common.OpenRestyWorkerProcesses = value
@@ -0,0 +1,231 @@
+package model
+
+import (
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/bwmarrin/snowflake"
+	"gorm.io/gorm"
+	"gorm.io/sharding"
+)
+
+const observabilityShardCount = 10
+
+var (
+	observabilityIDNode     *snowflake.Node
+	observabilityIDNodeErr  error
+	observabilityIDNodeOnce sync.Once
+)
+
+func registerSharding(db *gorm.DB, backend string) error {
+	if db == nil {
+		return nil
+	}
+	_ = backend
+	if err := db.Use(sharding.Register(sharding.Config{
+		ShardingKey:    "id",
+		NumberOfShards: observabilityShardCount,
+		ShardingAlgorithm: func(value any) (string, error) {
+			return observabilityShardSuffixForValue(value)
+		},
+		ShardingAlgorithmByPrimaryKey: func(id int64) string {
+			return observabilityShardSuffixForInt64(id)
+		},
+		PrimaryKeyGenerator: sharding.PKCustom,
+		PrimaryKeyGeneratorFn: func(tableIdx int64) int64 {
+			return 0
+		},
+	}, shardedObservabilityTables()...)); err != nil {
+		return fmt.Errorf("register observability sharding failed: %w", err)
+	}
+	return nil
+}
+
+func shardedObservabilityTables() []any {
+	return []any{
+		&NodeMetricSnapshot{},
+		&NodeRequestReport{},
+		&NodeAccessLog{},
+	}
+}
+
+func shardedObservabilityBaseTables() []string {
+	return []string{
+		"node_metric_snapshots",
+		"node_request_reports",
+		"node_access_logs",
+	}
+}
+
+func isShardedObservabilityTable(tableName string) bool {
+	switch strings.TrimSpace(tableName) {
+	case "node_metric_snapshots", "node_request_reports", "node_access_logs":
+		return true
+	default:
+		return false
+	}
+}
+
+func observabilityShardTables(baseTable string) []string {
+	tables := make([]string, 0, observabilityShardCount)
+	for _, suffix := range observabilityShardSuffixes() {
+		tables = append(tables, baseTable+suffix)
+	}
+	return tables
+}
+
+func observabilityShardSuffixes() []string {
+	suffixes := make([]string, 0, observabilityShardCount)
+	for index := 0; index < observabilityShardCount; index++ {
+		suffixes = append(suffixes, fmt.Sprintf("_%02d", index))
+	}
+	return suffixes
+}
+
+func observabilityShardSuffixForID(id uint) string {
+	return fmt.Sprintf("_%02d", uint64(id)%uint64(observabilityShardCount))
+}
+
+func observabilityShardSuffixForInt64(id int64) string {
+	if id < 0 {
+		id = -id
+	}
+	return fmt.Sprintf("_%02d", uint64(id)%uint64(observabilityShardCount))
+}
+
+func observabilityShardSuffixForValue(value any) (string, error) {
+	switch typed := value.(type) {
+	case int:
+		return observabilityShardSuffixForInt64(int64(typed)), nil
+	case int8:
+		return observabilityShardSuffixForInt64(int64(typed)), nil
+	case int16:
+		return observabilityShardSuffixForInt64(int64(typed)), nil
+	case int32:
+		return observabilityShardSuffixForInt64(int64(typed)), nil
+	case int64:
+		return observabilityShardSuffixForInt64(typed), nil
+	case uint:
+		return observabilityShardSuffixForID(typed), nil
+	case uint8:
+		return observabilityShardSuffixForID(uint(typed)), nil
+	case uint16:
+		return observabilityShardSuffixForID(uint(typed)), nil
+	case uint32:
+		return observabilityShardSuffixForID(uint(typed)), nil
+	case uint64:
+		return fmt.Sprintf("_%02d", typed%uint64(observabilityShardCount)), nil
+	case string:
+		id, err := strconv.ParseUint(strings.TrimSpace(typed), 10, 64)
+		if err != nil {
+			return "", fmt.Errorf("invalid sharding id %q", typed)
+		}
+		return fmt.Sprintf("_%02d", id%uint64(observabilityShardCount)), nil
+	default:
+		return "", fmt.Errorf("unsupported observability sharding value type %T", value)
+	}
+}
+
+func observabilityShardTableForID(baseTable string, id uint) string {
+	return baseTable + observabilityShardSuffixForID(id)
+}
+
+func legacyObservabilityShardTableName(tableName string) string {
+	return tableName + "_legacy_v2_to_v3"
+}
+
+func normalizeShardedDB(db *gorm.DB) *gorm.DB {
+	if db != nil {
+		return db
+	}
+	return DB
+}
+
+func sessionIgnoringSharding(db *gorm.DB) *gorm.DB {
+	db = normalizeShardedDB(db)
+	if db == nil {
+		return nil
+	}
+	return db.Session(&gorm.Session{}).Set(sharding.ShardingIgnoreStoreKey, true)
+}
+
+func baseDialector(db *gorm.DB) gorm.Dialector {
+	if db == nil {
+		return nil
+	}
+	if dialector, ok := db.Dialector.(sharding.ShardingDialector); ok {
+		return dialector.Dialector
+	}
+	return db.Dialector
+}
+
+func nextObservabilityID() (uint, error) {
+	observabilityIDNodeOnce.Do(func() {
+		observabilityIDNode, observabilityIDNodeErr = snowflake.NewNode(0)
+	})
+	if observabilityIDNodeErr != nil {
+		return 0, observabilityIDNodeErr
+	}
+	id := observabilityIDNode.Generate().Int64()
+	if id <= 0 {
+		return 0, fmt.Errorf("generated invalid observability id %d", id)
+	}
+	return uint(id), nil
+}
+
+func assignObservabilityID(id *uint) error {
+	if id == nil || *id != 0 {
+		return nil
+	}
+	generated, err := nextObservabilityID()
+	if err != nil {
+		return err
+	}
+	*id = generated
+	return nil
+}
+
+func queryAcrossShards[T any](baseTable string, query func(tx *gorm.DB) ([]T, error)) ([]T, error) {
+	return queryAcrossShardsWithDB(DB, baseTable, query)
+}
+
+func queryAcrossShardsWithDB[T any](db *gorm.DB, baseTable string, query func(tx *gorm.DB) ([]T, error)) ([]T, error) {
+	items := make([]T, 0)
+	db = normalizeShardedDB(db)
+	for _, table := range observabilityShardTables(baseTable) {
+		rows, err := query(db.Table(table))
+		if err != nil {
+			return nil, err
+		}
+		items = append(items, rows...)
+	}
+	return items, nil
+}
+
+func deleteAcrossShards(db *gorm.DB, baseTable string, model any, apply func(tx *gorm.DB) *gorm.DB) (int64, error) {
+	db = normalizeShardedDB(db)
+	var deleted int64
+	for _, table := range observabilityShardTables(baseTable) {
+		tx := db.Table(table)
+		if apply != nil {
+			tx = apply(tx)
+		} else {
+			tx = tx.Session(&gorm.Session{AllowGlobalUpdate: true})
+		}
+		result := tx.Delete(model)
+		if result.Error != nil {
+			return deleted, result.Error
+		}
+		deleted += result.RowsAffected
+	}
+	return deleted, nil
+}
+
+func sortShardRows[T any](items []T, less func(left T, right T) bool) {
+	sort.Slice(items, func(i int, j int) bool {
+		return less(items[i], items[j])
+	})
+}
@@ -55,6 +55,7 @@ func SetApiRouter(router *gin.Engine) {
 			optionRoute.GET("/", controller.GetOptions)
 			optionRoute.POST("/update", controller.UpdateOption)
 			optionRoute.POST("/geoip/lookup", controller.LookupGeoIP)
+			optionRoute.POST("/database/cleanup", controller.CleanupDatabaseObservability)
 		}
 		updateRoute := apiRouter.Group("/update")
 		updateRoute.Use(middleware.RootAuth(), middleware.NoTokenAuth())
@@ -2,6 +2,7 @@ package service

 import (
 	"openflare/model"
+	"strings"
 	"testing"
 	"time"
 )
@@ -60,9 +61,7 @@ func TestListAccessLogsIncludesSummaryTotals(t *testing.T) {
 			StatusCode: 200,
 		},
 	}
-	if err := model.DB.Create(&logs).Error; err != nil {
-		t.Fatalf("failed to seed access logs: %v", err)
-	}
+	seedNodeAccessLogs(t, logs)

 	result, err := ListAccessLogs(AccessLogQuery{Page: 0, PageSize: 2})
 	if err != nil {
@@ -121,9 +120,7 @@ func TestListAccessLogsUsesDefaultPageSize(t *testing.T) {
 			StatusCode: 200,
 		})
 	}
-	if err := model.DB.Create(&logs).Error; err != nil {
-		t.Fatalf("failed to seed access logs: %v", err)
-	}
+	seedNodeAccessLogs(t, logs)

 	result, err := ListAccessLogs(AccessLogQuery{})
 	if err != nil {
@@ -143,7 +140,7 @@ func TestListAccessLogsUsesDefaultPageSize(t *testing.T) {
 func TestListFoldedAccessLogsAndIPSummaries(t *testing.T) {
 	setupServiceTestDB(t)

-	now := time.Now().UTC()
+	now := time.Date(2026, 3, 19, 8, 12, 30, 0, time.UTC)
 	if err := model.DB.Create(&model.Node{
 		NodeID: "node-folded",
 		Name:   "edge-folded",
@@ -176,9 +173,7 @@ func TestListFoldedAccessLogsAndIPSummaries(t *testing.T) {
 			StatusCode: 404,
 		},
 	}
-	if err := model.DB.Create(&logs).Error; err != nil {
-		t.Fatalf("failed to seed access logs: %v", err)
-	}
+	seedNodeAccessLogs(t, logs)

 	folded, err := ListFoldedAccessLogs(AccessLogQuery{
 		NodeID:      "node-folded",
@@ -221,7 +216,7 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) {
 	setupServiceTestDB(t)

 	now := time.Now().UTC()
-	if err := model.DB.Create([]*model.NodeAccessLog{
+	seedNodeAccessLogs(t, []*model.NodeAccessLog{
 		{
 			NodeID:     "node-cleanup",
 			LoggedAt:   now.Add(-10 * 24 * time.Hour),
@@ -238,9 +233,7 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) {
 			Path:       "/recent",
 			StatusCode: 200,
 		},
-	}).Error; err != nil {
-		t.Fatalf("failed to seed cleanup logs: %v", err)
-	}
+	})

 	result, err := CleanupAccessLogs(AccessLogCleanupInput{RetentionDays: 7})
 	if err != nil {
@@ -258,3 +251,45 @@ func TestCleanupAccessLogsDeletesExpiredData(t *testing.T) {
 		t.Fatalf("unexpected remaining logs after cleanup: %+v", remaining.Items)
 	}
 }
+
+func TestPersistNodeAccessLogsTruncatesLongPath(t *testing.T) {
+	setupServiceTestDB(t)
+
+	longPath := "/" + strings.Repeat("a", 140)
+	reportedAt := time.Now().UTC()
+	if err := persistNodeAccessLogs(model.DB, "node-truncate", []AgentNodeAccessLog{
+		{
+			LoggedAtUnix: reportedAt.Unix(),
+			RemoteAddr:   "203.0.113.10",
+			Host:         "truncate.example.com",
+			Path:         longPath,
+			StatusCode:   200,
+		},
+	}, reportedAt); err != nil {
+		t.Fatalf("persistNodeAccessLogs failed: %v", err)
+	}
+
+	logs, err := model.ListNodeAccessLogs(model.NodeAccessLogQuery{
+		NodeID:   "node-truncate",
+		Page:     0,
+		PageSize: 10,
+	})
+	if err != nil {
+		t.Fatalf("ListNodeAccessLogs failed: %v", err)
+	}
+	if len(logs) != 1 {
+		t.Fatalf("expected one stored log, got %+v", logs)
+	}
+	if got := len([]rune(logs[0].Path)); got != nodeAccessLogPathMaxLength {
+		t.Fatalf("expected truncated path length %d, got %d (%q)", nodeAccessLogPathMaxLength, got, logs[0].Path)
+	}
+}
+
+func seedNodeAccessLogs(t *testing.T, logs []*model.NodeAccessLog) {
+	t.Helper()
+	for _, item := range logs {
+		if err := model.DB.Create(item).Error; err != nil {
+			t.Fatalf("failed to seed access log: %v", err)
+		}
+	}
+}
@@ -0,0 +1,182 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"openflare/common"
+	"openflare/model"
+	"strings"
+	"time"
+)
+
+const (
+	DatabaseCleanupTargetAccessLogs      = "node_access_logs"
+	DatabaseCleanupTargetMetricSnapshots = "node_metric_snapshots"
+	DatabaseCleanupTargetRequestReports  = "node_request_reports"
+)
+
+var databaseCleanupTargets = map[string]string{
+	DatabaseCleanupTargetAccessLogs:      "访问日志",
+	DatabaseCleanupTargetMetricSnapshots: "性能快照",
+	DatabaseCleanupTargetRequestReports:  "请求聚合",
+}
+
+type DatabaseCleanupInput struct {
+	Target        string `json:"target"`
+	RetentionDays *int   `json:"retention_days"`
+}
+
+type DatabaseCleanupResult struct {
+	Target        string     `json:"target"`
+	TargetLabel   string     `json:"target_label"`
+	DeletedCount  int64      `json:"deleted_count"`
+	DeleteAll     bool       `json:"delete_all"`
+	RetentionDays *int       `json:"retention_days,omitempty"`
+	Cutoff        *time.Time `json:"cutoff,omitempty"`
+}
+
+type DatabaseAutoCleanupSummary struct {
+	RetentionDays int                     `json:"retention_days"`
+	ExecutedAt    time.Time               `json:"executed_at"`
+	Results       []DatabaseCleanupResult `json:"results"`
+}
+
+func CleanupDatabaseObservability(input DatabaseCleanupInput) (*DatabaseCleanupResult, error) {
+	target := strings.TrimSpace(input.Target)
+	targetLabel, ok := databaseCleanupTargets[target]
+	if !ok {
+		return nil, errors.New("unsupported cleanup target")
+	}
+	if input.RetentionDays != nil && *input.RetentionDays <= 0 {
+		return nil, errors.New("retention_days 必须为大于 0 的整数")
+	}
+
+	result := &DatabaseCleanupResult{
+		Target:      target,
+		TargetLabel: targetLabel,
+		DeleteAll:   input.RetentionDays == nil,
+	}
+
+	if input.RetentionDays == nil {
+		deleted, err := deleteAllObservabilityRows(target)
+		if err != nil {
+			return nil, err
+		}
+		result.DeletedCount = deleted
+		return result, nil
+	}
+
+	retentionDays := *input.RetentionDays
+	cutoff := time.Now().UTC().Add(-time.Duration(retentionDays) * 24 * time.Hour)
+	deleted, err := deleteObservabilityRowsBefore(target, cutoff)
+	if err != nil {
+		return nil, err
+	}
+	result.DeletedCount = deleted
+	result.RetentionDays = &retentionDays
+	result.Cutoff = &cutoff
+	return result, nil
+}
+
+func RunDatabaseAutoCleanupOnce(now time.Time) (*DatabaseAutoCleanupSummary, error) {
+	if !common.DatabaseAutoCleanupEnabled {
+		return nil, nil
+	}
+	if common.DatabaseAutoCleanupRetentionDays < 1 {
+		return nil, fmt.Errorf("database auto cleanup retention_days must be at least 1")
+	}
+
+	retentionDays := common.DatabaseAutoCleanupRetentionDays
+	results := make([]DatabaseCleanupResult, 0, len(databaseCleanupTargets))
+	for _, target := range []string{
+		DatabaseCleanupTargetAccessLogs,
+		DatabaseCleanupTargetMetricSnapshots,
+		DatabaseCleanupTargetRequestReports,
+	} {
+		result, err := CleanupDatabaseObservability(DatabaseCleanupInput{
+			Target:        target,
+			RetentionDays: &retentionDays,
+		})
+		if err != nil {
+			return nil, err
+		}
+		results = append(results, *result)
+	}
+
+	return &DatabaseAutoCleanupSummary{
+		RetentionDays: retentionDays,
+		ExecutedAt:    now.UTC(),
+		Results:       results,
+	}, nil
+}
+
+func StartDatabaseAutoCleanupScheduler(ctx context.Context) {
+	go func() {
+		for {
+			wait := time.Until(nextDatabaseAutoCleanupTime(time.Now()))
+			timer := time.NewTimer(wait)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
+
+			summary, err := RunDatabaseAutoCleanupOnce(time.Now())
+			if err != nil {
+				slog.Error("database auto cleanup failed", "error", err)
+				continue
+			}
+			if summary == nil {
+				continue
+			}
+			totalDeleted := int64(0)
+			for _, item := range summary.Results {
+				totalDeleted += item.DeletedCount
+			}
+			slog.Info(
+				"database auto cleanup completed",
+				"retention_days",
+				summary.RetentionDays,
+				"deleted_count",
+				totalDeleted,
+			)
+		}
+	}()
+}
+
+func nextDatabaseAutoCleanupTime(now time.Time) time.Time {
+	next := time.Date(now.Year(), now.Month(), now.Day(), 3, 0, 0, 0, now.Location())
+	if !next.After(now) {
+		next = next.Add(24 * time.Hour)
+	}
+	return next
+}
+
+func deleteAllObservabilityRows(target string) (int64, error) {
+	switch target {
+	case DatabaseCleanupTargetAccessLogs:
+		return model.DeleteAllNodeAccessLogs(nil)
+	case DatabaseCleanupTargetMetricSnapshots:
+		return model.DeleteAllNodeMetricSnapshots(nil)
+	case DatabaseCleanupTargetRequestReports:
+		return model.DeleteAllNodeRequestReports(nil)
+	default:
+		return 0, errors.New("unsupported cleanup target")
+	}
+}
+
+func deleteObservabilityRowsBefore(target string, cutoff time.Time) (int64, error) {
+	switch target {
+	case DatabaseCleanupTargetAccessLogs:
+		return model.DeleteNodeAccessLogsBefore(cutoff)
+	case DatabaseCleanupTargetMetricSnapshots:
+		return model.DeleteNodeMetricSnapshotsBefore(nil, cutoff)
+	case DatabaseCleanupTargetRequestReports:
+		return model.DeleteNodeRequestReportsBefore(nil, cutoff)
+	default:
+		return 0, errors.New("unsupported cleanup target")
+	}
+}
@@ -0,0 +1,165 @@
+package service
+
+import (
+	"openflare/common"
+	"openflare/model"
+	"testing"
+	"time"
+)
+
+func TestCleanupDatabaseObservabilityDeletesTargetedRows(t *testing.T) {
+	setupServiceTestDB(t)
+
+	now := time.Now().UTC()
+	if err := model.DB.Create(&model.NodeMetricSnapshot{
+		NodeID:          "node-a",
+		CapturedAt:      now.Add(-10 * 24 * time.Hour),
+		CPUUsagePercent: 10,
+	}).Error; err != nil {
+		t.Fatalf("seed old metric snapshot: %v", err)
+	}
+	if err := model.DB.Create(&model.NodeMetricSnapshot{
+		NodeID:          "node-a",
+		CapturedAt:      now.Add(-12 * time.Hour),
+		CPUUsagePercent: 20,
+	}).Error; err != nil {
+		t.Fatalf("seed recent metric snapshot: %v", err)
+	}
+
+	retentionDays := 7
+	result, err := CleanupDatabaseObservability(DatabaseCleanupInput{
+		Target:        DatabaseCleanupTargetMetricSnapshots,
+		RetentionDays: &retentionDays,
+	})
+	if err != nil {
+		t.Fatalf("CleanupDatabaseObservability failed: %v", err)
+	}
+	if result.DeleteAll {
+		t.Fatal("expected retention cleanup instead of delete_all")
+	}
+	if result.DeletedCount != 1 {
+		t.Fatalf("expected 1 deleted row, got %+v", result)
+	}
+
+	rows, err := model.ListMetricSnapshotsSince(time.Time{})
+	if err != nil {
+		t.Fatalf("ListMetricSnapshotsSince failed: %v", err)
+	}
+	if len(rows) != 1 || rows[0].CPUUsagePercent != 20 {
+		t.Fatalf("unexpected remaining metric snapshots: %+v", rows)
+	}
+}
+
+func TestCleanupDatabaseObservabilityDeletesAllRowsWhenRetentionMissing(t *testing.T) {
+	setupServiceTestDB(t)
+
+	now := time.Now().UTC()
+	if err := model.DB.Create(&model.NodeAccessLog{
+		NodeID:     "node-a",
+		LoggedAt:   now.Add(-3 * time.Hour),
+		RemoteAddr: "203.0.113.1",
+		Host:       "example.com",
+		Path:       "/one",
+		StatusCode: 200,
+	}).Error; err != nil {
+		t.Fatalf("seed first access log: %v", err)
+	}
+	if err := model.DB.Create(&model.NodeAccessLog{
+		NodeID:     "node-a",
+		LoggedAt:   now.Add(-2 * time.Hour),
+		RemoteAddr: "203.0.113.2",
+		Host:       "example.com",
+		Path:       "/two",
+		StatusCode: 502,
+	}).Error; err != nil {
+		t.Fatalf("seed second access log: %v", err)
+	}
+
+	result, err := CleanupDatabaseObservability(DatabaseCleanupInput{
+		Target: DatabaseCleanupTargetAccessLogs,
+	})
+	if err != nil {
+		t.Fatalf("CleanupDatabaseObservability failed: %v", err)
+	}
+	if !result.DeleteAll || result.DeletedCount != 2 {
+		t.Fatalf("unexpected delete-all result: %+v", result)
+	}
+
+	rows, err := model.ListNodeAccessLogs(model.NodeAccessLogQuery{Page: 0, PageSize: 10})
+	if err != nil {
+		t.Fatalf("ListNodeAccessLogs failed: %v", err)
+	}
+	if len(rows) != 0 {
+		t.Fatalf("expected all access logs deleted, got %+v", rows)
+	}
+}
+
+func TestRunDatabaseAutoCleanupOnceDeletesAllObservabilityTargets(t *testing.T) {
+	setupServiceTestDB(t)
+
+	now := time.Now().UTC()
+	if err := model.DB.Create(&model.NodeAccessLog{
+		NodeID:     "node-a",
+		LoggedAt:   now.Add(-48 * time.Hour),
+		RemoteAddr: "203.0.113.10",
+		Host:       "example.com",
+		Path:       "/access",
+		StatusCode: 200,
+	}).Error; err != nil {
+		t.Fatalf("seed access log: %v", err)
+	}
+	if err := model.DB.Create(&model.NodeMetricSnapshot{
+		NodeID:          "node-a",
+		CapturedAt:      now.Add(-48 * time.Hour),
+		CPUUsagePercent: 10,
+	}).Error; err != nil {
+		t.Fatalf("seed metric snapshot: %v", err)
+	}
+	if err := model.DB.Create(&model.NodeRequestReport{
+		NodeID:          "node-a",
+		WindowStartedAt: now.Add(-49 * time.Hour),
+		WindowEndedAt:   now.Add(-48 * time.Hour),
+		RequestCount:    15,
+	}).Error; err != nil {
+		t.Fatalf("seed request report: %v", err)
+	}
+
+	previousEnabled := common.DatabaseAutoCleanupEnabled
+	previousRetentionDays := common.DatabaseAutoCleanupRetentionDays
+	common.DatabaseAutoCleanupEnabled = true
+	common.DatabaseAutoCleanupRetentionDays = 1
+	t.Cleanup(func() {
+		common.DatabaseAutoCleanupEnabled = previousEnabled
+		common.DatabaseAutoCleanupRetentionDays = previousRetentionDays
+	})
+
+	summary, err := RunDatabaseAutoCleanupOnce(now)
+	if err != nil {
+		t.Fatalf("RunDatabaseAutoCleanupOnce failed: %v", err)
+	}
+	if summary == nil || len(summary.Results) != 3 {
+		t.Fatalf("unexpected auto cleanup summary: %+v", summary)
+	}
+
+	accessLogs, err := model.ListNodeAccessLogs(model.NodeAccessLogQuery{Page: 0, PageSize: 10})
+	if err != nil {
+		t.Fatalf("ListNodeAccessLogs failed: %v", err)
+	}
+	if len(accessLogs) != 0 {
+		t.Fatalf("expected auto cleanup to delete access logs, got %+v", accessLogs)
+	}
+	metricSnapshots, err := model.ListMetricSnapshotsSince(time.Time{})
+	if err != nil {
+		t.Fatalf("ListMetricSnapshotsSince failed: %v", err)
+	}
+	if len(metricSnapshots) != 0 {
+		t.Fatalf("expected auto cleanup to delete metric snapshots, got %+v", metricSnapshots)
+	}
+	requestReports, err := model.ListRequestReportsSince(time.Time{})
+	if err != nil {
+		t.Fatalf("ListRequestReportsSince failed: %v", err)
+	}
+	if len(requestReports) != 0 {
+		t.Fatalf("expected auto cleanup to delete request reports, got %+v", requestReports)
+	}
+}
@@ -1,6 +1,7 @@
 package service

 import (
+	"encoding/json"
 	"io"
 	"net"
 	"net/http"
@@ -677,6 +678,9 @@ func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) {
 				Severity:        NodeHealthSeverityCritical,
 				Message:         "reload failed",
 				TriggeredAtUnix: time.Now().Add(-2 * time.Minute).Unix(),
+				Metadata: map[string]string{
+					"source": "runtime",
+				},
 			},
 		},
 	})
@@ -731,6 +735,16 @@ func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) {
 	if len(events) != 1 || events[0].EventType != "openresty_unhealthy" {
 		t.Fatalf("unexpected active health events: %+v", events)
 	}
+	if events[0].MetadataJSON == "" {
+		t.Fatal("expected metadata_json to persist")
+	}
+	var metadata map[string]string
+	if err := json.Unmarshal([]byte(events[0].MetadataJSON), &metadata); err != nil {
+		t.Fatalf("expected metadata_json to be valid json: %v", err)
+	}
+	if metadata["source"] != "runtime" {
+		t.Fatalf("unexpected metadata json: %+v", metadata)
+	}
 }

 func TestHeartbeatNodePersistsBufferedObservabilityPayload(t *testing.T) {
@@ -868,6 +882,15 @@ func TestHeartbeatNodePersistsBufferedObservabilityPayload(t *testing.T) {
 					TopDomains:          map[string]int64{"edge.example.com": 40},
 					SourceCountries:     map[string]int64{"CN": 20},
 				},
+				AccessLogs: []AgentNodeAccessLog{
+					{
+						LoggedAtUnix: now.Add(-110 * time.Second).Unix(),
+						RemoteAddr:   "203.0.113.21",
+						Host:         "edge.example.com",
+						Path:         "/buffered",
+						StatusCode:   200,
+					},
+				},
 			},
 		},
 	})
@@ -889,6 +912,18 @@ func TestHeartbeatNodePersistsBufferedObservabilityPayload(t *testing.T) {
 	if len(reports) != 2 {
 		t.Fatalf("expected replay dedupe to keep report count stable, got %+v", reports)
 	}
+	accessLogs, err = model.ListNodeAccessLogs(model.NodeAccessLogQuery{
+		NodeID:   node.NodeID,
+		Since:    time.Time{},
+		Page:     0,
+		PageSize: 10,
+	})
+	if err != nil {
+		t.Fatalf("expected node access logs query to succeed after replay: %v", err)
+	}
+	if len(accessLogs) != 1 {
+		t.Fatalf("expected replay dedupe to keep access log count stable, got %+v", accessLogs)
+	}
 }

 func TestListAccessLogsUsesPagination(t *testing.T) {
@@ -908,7 +943,7 @@ func TestListAccessLogsUsesPagination(t *testing.T) {
 	}

 	now := time.Now().UTC()
-	if err := model.DB.Create([]*model.NodeAccessLog{
+	for _, item := range []*model.NodeAccessLog{
 		{
 			NodeID:     node.NodeID,
 			LoggedAt:   now.Add(-10 * time.Second),
@@ -936,8 +971,10 @@ func TestListAccessLogsUsesPagination(t *testing.T) {
 			Path:       "/three",
 			StatusCode: 502,
 		},
-	}).Error; err != nil {
-		t.Fatalf("failed to seed access logs: %v", err)
+	} {
+		if err := model.DB.Create(item).Error; err != nil {
+			t.Fatalf("failed to seed access logs: %v", err)
+		}
 	}

 	pageOne, err := ListAccessLogs(AccessLogQuery{
@@ -1404,7 +1441,7 @@ func TestGetDashboardOverview(t *testing.T) {
 	}).Insert(); err != nil {
 		t.Fatalf("failed to insert node b traffic report: %v", err)
 	}
-	if err := model.DB.Create([]*model.NodeAccessLog{
+	for _, item := range []*model.NodeAccessLog{
 		{
 			NodeID:     "node-dashboard-a",
 			LoggedAt:   now.Add(-30 * time.Minute),
@@ -1432,8 +1469,10 @@ func TestGetDashboardOverview(t *testing.T) {
 			Path:       "/edge",
 			StatusCode: 502,
 		},
-	}).Error; err != nil {
-		t.Fatalf("failed to seed dashboard access logs: %v", err)
+	} {
+		if err := model.DB.Create(item).Error; err != nil {
+			t.Fatalf("failed to seed dashboard access logs: %v", err)
+		}
 	}

 	if err := model.DB.Create(&model.NodeHealthEvent{
@@ -18,6 +18,7 @@ const (
 	NodeHealthSeverityWarning     = "warning"
 	NodeHealthSeverityCritical    = "critical"
 	nodeAccessLogRetentionWindow  = nodeAccessLogRetentionDays * 24 * time.Hour
+	nodeAccessLogPathMaxLength    = 100
 )

 type AgentNodeSystemProfile struct {
@@ -151,7 +152,6 @@ func persistNodeSystemProfile(tx *gorm.DB, nodeID string, profile *AgentNodeSyst
 		TotalDiskBytes:   profile.TotalDiskBytes,
 		UptimeSeconds:    profile.UptimeSeconds,
 		ReportedAt:       timeFromUnix(profile.ReportedAtUnix, reportedAt),
-		RawJSON:          marshalJSON(profile),
 	}
 	return tx.Model(&model.NodeSystemProfile{}).Where("node_id = ?", nodeID).Assign(record).FirstOrCreate(record).Error
 }
@@ -175,9 +175,15 @@ func persistNodeMetricSnapshot(tx *gorm.DB, nodeID string, snapshot *AgentNodeMe
 		OpenrestyRxBytes:     snapshot.OpenrestyRxBytes,
 		OpenrestyTxBytes:     snapshot.OpenrestyTxBytes,
 		OpenrestyConnections: snapshot.OpenrestyConnections,
-		RawJSON:              marshalJSON(snapshot),
 	}
-	return tx.Where("node_id = ? AND captured_at = ?", nodeID, record.CapturedAt).Assign(record).FirstOrCreate(record).Error
+	exists, err := model.NodeMetricSnapshotExists(tx, nodeID, record.CapturedAt)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return nil
+	}
+	return tx.Create(record).Error
 }

 func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *AgentNodeTrafficReport, reportedAt time.Time) error {
@@ -197,9 +203,15 @@ func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *AgentNodeTraff
 		StatusCodesJSON:     marshalJSON(report.StatusCodes),
 		TopDomainsJSON:      marshalJSON(report.TopDomains),
 		SourceCountriesJSON: marshalJSON(report.SourceCountries),
-		RawJSON:             marshalJSON(report),
 	}
-	return tx.Where("node_id = ? AND window_started_at = ? AND window_ended_at = ?", nodeID, record.WindowStartedAt, record.WindowEndedAt).Assign(record).FirstOrCreate(record).Error
+	exists, err := model.NodeRequestReportExists(tx, nodeID, record.WindowStartedAt, record.WindowEndedAt)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return nil
+	}
+	return tx.Create(record).Error
 }

 func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog, reportedAt time.Time) error {
@@ -220,26 +232,25 @@ func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog
 			RemoteAddr: strings.TrimSpace(item.RemoteAddr),
 			Region:     "",
 			Host:       strings.TrimSpace(item.Host),
-			Path:       strings.TrimSpace(item.Path),
+			Path:       truncateForDatabase(strings.TrimSpace(item.Path), nodeAccessLogPathMaxLength),
 			StatusCode: item.StatusCode,
-			RawJSON:    marshalJSON(item),
 		}
 		if resolver != nil {
 			record.Region = resolver.Resolve(record.RemoteAddr)
 		}
-		if err := tx.Where(
-			"node_id = ? AND logged_at = ? AND remote_addr = ? AND host = ? AND path = ? AND status_code = ?",
-			nodeID,
-			record.LoggedAt,
-			record.RemoteAddr,
-			record.Host,
-			record.Path,
-			record.StatusCode,
-		).Assign(record).FirstOrCreate(record).Error; err != nil {
+		exists, err := model.NodeAccessLogExists(tx, record)
+		if err != nil {
+			return err
+		}
+		if exists {
+			continue
+		}
+		if err := tx.Create(record).Error; err != nil {
 			return err
 		}
 	}
-	return tx.Where("node_id = ? AND logged_at < ?", nodeID, reportedAt.Add(-nodeAccessLogRetentionWindow)).Delete(&model.NodeAccessLog{}).Error
+	_, err = model.DeleteNodeAccessLogsByNodeBefore(tx, nodeID, reportedAt.Add(-nodeAccessLogRetentionWindow))
+	return err
 }

 func reconcileNodeHealthEvents(tx *gorm.DB, nodeID string, events []AgentNodeHealthEvent, reportedAt time.Time) error {
@@ -274,7 +285,7 @@ func reconcileNodeHealthEvents(tx *gorm.DB, nodeID string, events []AgentNodeHea
 			existing.Message = normalizeHealthEventMessage(event.Message)
 			existing.LastTriggeredAt = triggeredAt
 			existing.ReportedAt = reportedAt
-			existing.RawJSON = marshalJSON(event)
+			existing.MetadataJSON = marshalJSON(event.Metadata)
 			existing.ResolvedAt = nil
 			if err := tx.Save(existing).Error; err != nil {
 				return err
@@ -290,7 +301,7 @@ func reconcileNodeHealthEvents(tx *gorm.DB, nodeID string, events []AgentNodeHea
 			FirstTriggeredAt: triggeredAt,
 			LastTriggeredAt:  triggeredAt,
 			ReportedAt:       reportedAt,
-			RawJSON:          marshalJSON(event),
+			MetadataJSON:     marshalJSON(event.Metadata),
 		}
 		if err := tx.Create(record).Error; err != nil {
 			return err
@@ -183,16 +183,17 @@ export interface NodeObservabilityTrends {
  disk_io_24h: NodeDiskIOTrendPoint[];
 }

-export interface NodeHealthEvent {
-  event_type: string;
-  severity: string;
-  status: string;
-  message: string;
-  first_triggered_at: string;
-  last_triggered_at: string;
-  reported_at: string;
-  resolved_at?: string | null;
-}
+export interface NodeHealthEvent {
+  event_type: string;
+  severity: string;
+  status: string;
+  message: string;
+  metadata_json?: string;
+  first_triggered_at: string;
+  last_triggered_at: string;
+  reported_at: string;
+  resolved_at?: string | null;
+}

 export interface NodeObservability {
  node_id: string;
@@ -2,6 +2,8 @@ import { apiRequest } from '@/lib/api/client';

 import type {
  BootstrapTokenPayload,
+  DatabaseCleanupPayload,
+  DatabaseCleanupResult,
  GeoIPLookupResult,
  OptionItem,
  SettingsProfile,
@@ -26,6 +28,13 @@ export function lookupGeoIP(provider: string, ip: string) {
  });
 }

+export function cleanupDatabaseObservability(payload: DatabaseCleanupPayload) {
+  return apiRequest<DatabaseCleanupResult>('/option/database/cleanup', {
+    method: 'POST',
+    body: JSON.stringify(payload),
+  });
+}
+
 export function getBootstrapToken() {
  return apiRequest<BootstrapTokenPayload>('/nodes/bootstrap-token');
 }
@@ -7,6 +7,7 @@ import { EmptyState } from '@/components/feedback/empty-state';
 import { ErrorState } from '@/components/feedback/error-state';
 import { InlineMessage } from '@/components/feedback/inline-message';
 import { LoadingState } from '@/components/feedback/loading-state';
+import { AppModal } from '@/components/ui/app-modal';
 import { TurnstileWidget } from '@/components/forms/turnstile-widget';
 import { useAuth } from '@/components/providers/auth-provider';
 import { PageHeader } from '@/components/layout/page-header';
@@ -17,6 +18,7 @@ import { getPublicStatus } from '@/features/auth/api/public';
 import {
  bindEmail,
  bindWeChat,
+  cleanupDatabaseObservability,
  generateAccessToken,
  getBootstrapToken,
  getOptions,
@@ -28,12 +30,15 @@ import {
 } from '@/features/settings/api/settings';
 import type {
  BootstrapTokenPayload,
+  DatabaseCleanupResult,
+  DatabaseCleanupTarget,
  GeoIPLookupResult,
  OptionItem,
  UpdateSelfPayload,
 } from '@/features/settings/types';
 import {
  CodeBlock,
+  DangerButton,
  PrimaryButton,
  ResourceField,
  ResourceInput,
@@ -126,6 +131,11 @@ const defaultOtherFields = {
  Footer: '',
 };

+const defaultDatabaseFields = {
+  DatabaseAutoCleanupEnabled: false,
+  DatabaseAutoCleanupRetentionDays: '30',
+};
+
 const defaultProfileFields: UpdateSelfPayload = {
  username: '',
  display_name: '',
@@ -137,7 +147,17 @@ type FeedbackState = {
  message: string;
 };

-type SettingsTab = 'personal' | 'operation' | 'system' | 'other';
+type CleanupModalState = {
+  target: DatabaseCleanupTarget;
+  label: string;
+};
+
+type SettingsTab =
+  | 'personal'
+  | 'operation'
+  | 'database'
+  | 'system'
+  | 'other';

 function getErrorMessage(error: unknown) {
  return error instanceof Error ? error.message : '请求失败，请稍后重试。';
@@ -227,12 +247,16 @@ export function SettingsPage() {
    defaultOperationFields,
  );
  const [otherFields, setOtherFields] = useState(defaultOtherFields);
+  const [databaseFields, setDatabaseFields] = useState(defaultDatabaseFields);
  const [accessToken, setAccessToken] = useState('');
  const [wechatCode, setWeChatCode] = useState('');
  const [emailAddress, setEmailAddress] = useState('');
  const [emailCode, setEmailCode] = useState('');
  const [emailTurnstileToken, setEmailTurnstileToken] = useState('');
  const [geoIPTestIP, setGeoIPTestIP] = useState('8.8.8.8');
+  const [cleanupModalState, setCleanupModalState] =
+    useState<CleanupModalState | null>(null);
+  const [cleanupRetentionDays, setCleanupRetentionDays] = useState('');

  const isRoot = (user?.role ?? 0) >= 100;

@@ -410,6 +434,14 @@ export function SettingsPage() {
      About: optionMap.About ?? '',
      Footer: optionMap.Footer ?? '',
    });
+    setDatabaseFields({
+      DatabaseAutoCleanupEnabled: toBoolean(
+        optionMap.DatabaseAutoCleanupEnabled,
+        false,
+      ),
+      DatabaseAutoCleanupRetentionDays:
+        optionMap.DatabaseAutoCleanupRetentionDays ?? '30',
+    });
  }, [optionsQuery.data, publicStatusQuery.data?.server_address]);

  const rotateTokenMutation = useMutation({
@@ -451,6 +483,23 @@ export function SettingsPage() {
      lookupGeoIP(provider, ip),
  });

+  const databaseCleanupMutation = useMutation({
+    mutationFn: cleanupDatabaseObservability,
+    onSuccess: (result: DatabaseCleanupResult) => {
+      setCleanupModalState(null);
+      setCleanupRetentionDays('');
+      setFeedback({
+        tone: 'success',
+        message: result.delete_all
+          ? `已清空${result.target_label}数据，共删除 ${result.deleted_count} 条。`
+          : `已清理${result.target_label}中超出保留期的数据，共删除 ${result.deleted_count} 条。`,
+      });
+    },
+    onError: (error) => {
+      setFeedback({ tone: 'danger', message: getErrorMessage(error) });
+    },
+  });
+
  const discoveryToken = bootstrapQuery.data?.discovery_token ?? '';
  const discoveryCommand =
    isRoot && operationFields.ServerAddress && discoveryToken
@@ -476,6 +525,11 @@ export function SettingsPage() {
              label: '系统设置',
              description: '登录注册、SMTP、OAuth、限流与风控开关。',
            },
+            {
+              key: 'database' as const,
+              label: '数据库',
+              description: '观测数据清理与每日自动保留策略。',
+            },
            {
              key: 'other' as const,
              label: '其他设置',
@@ -1270,6 +1324,168 @@ export function SettingsPage() {
      );
    }

+    if (activeTab === 'database') {
+      return (
+        <div className="grid gap-6 xl:grid-cols-2 xl:items-start">
+          <AppCard
+            title="自动数据清理"
+            description="每天凌晨 3 点自动清理超出保留期的观测数据，统一作用于访问日志、性能快照和请求聚合。"
+            action={
+              <PrimaryButton
+                type="button"
+                onClick={() =>
+                  void runBusyAction('database-auto-cleanup', async () => {
+                    const retentionDays = Number.parseInt(
+                      databaseFields.DatabaseAutoCleanupRetentionDays,
+                      10,
+                    );
+                    if (Number.isNaN(retentionDays) || retentionDays < 1) {
+                      throw new Error('自动清理保留天数至少为 1 天。');
+                    }
+                    await saveOptionEntries(
+                      [
+                        [
+                          'DatabaseAutoCleanupEnabled',
+                          String(databaseFields.DatabaseAutoCleanupEnabled),
+                        ],
+                        [
+                          'DatabaseAutoCleanupRetentionDays',
+                          String(retentionDays),
+                        ],
+                      ],
+                      '数据库自动清理设置已保存。',
+                    );
+                  })
+                }
+                disabled={busyKey === 'database-auto-cleanup'}
+              >
+                {busyKey === 'database-auto-cleanup'
+                  ? '保存中...'
+                  : '保存自动清理'}
+              </PrimaryButton>
+            }
+          >
+            <div className="rounded-2xl border border-[var(--border-default)] bg-[var(--surface-elevated)] p-5">
+              <div className="space-y-5">
+                <ToggleField
+                  label="启用每日自动清理"
+                  description="开启后，服务端每天自动删除保留天数之外的观测数据。"
+                  checked={databaseFields.DatabaseAutoCleanupEnabled}
+                  onChange={(checked) =>
+                    setDatabaseFields((previous) => ({
+                      ...previous,
+                      DatabaseAutoCleanupEnabled: checked,
+                    }))
+                  }
+                />
+                <div className="border-t border-[var(--border-default)] pt-5">
+                  <ResourceField
+                    label="自动清理保留天数"
+                    hint="必须至少保留 1 天，服务端不允许配置为 24 小时以内。"
+                  >
+                    <ResourceInput
+                      type="number"
+                      min={1}
+                      value={databaseFields.DatabaseAutoCleanupRetentionDays}
+                      onChange={(event) =>
+                        setDatabaseFields((previous) => ({
+                          ...previous,
+                          DatabaseAutoCleanupRetentionDays: event.target.value,
+                        }))
+                      }
+                      placeholder="例如 30"
+                    />
+                  </ResourceField>
+                  <div className="mt-4 grid gap-4 md:grid-cols-3">
+                    <div className="rounded-2xl border border-[var(--border-default)] bg-[var(--surface-base)] px-4 py-4">
+                      <p className="text-xs tracking-[0.2em] uppercase text-[var(--foreground-muted)]">
+                        触发频率
+                      </p>
+                      <p className="mt-2 text-sm font-semibold text-[var(--foreground-primary)]">
+                        每天一次
+                      </p>
+                    </div>
+                    <div className="rounded-2xl border border-[var(--border-default)] bg-[var(--surface-base)] px-4 py-4">
+                      <p className="text-xs tracking-[0.2em] uppercase text-[var(--foreground-muted)]">
+                        默认执行时间
+                      </p>
+                      <p className="mt-2 text-sm font-semibold text-[var(--foreground-primary)]">
+                        凌晨 3:00
+                      </p>
+                    </div>
+                    <div className="rounded-2xl border border-[var(--border-default)] bg-[var(--surface-base)] px-4 py-4">
+                      <p className="text-xs tracking-[0.2em] uppercase text-[var(--foreground-muted)]">
+                        生效范围
+                      </p>
+                      <p className="mt-2 text-sm font-semibold text-[var(--foreground-primary)]">
+                        三类观测表
+                      </p>
+                    </div>
+                  </div>
+                </div>
+              </div>
+            </div>
+          </AppCard>
+
+          <AppCard
+            title="数据清理"
+            description="用于手动清理单类观测数据。保留天数留空时会直接删除该类数据的全部历史记录。"
+          >
+            <div className="grid gap-5 xl:grid-cols-3">
+              {[
+                {
+                  target: 'node_access_logs' as const,
+                  label: '访问日志',
+                  description:
+                    '清理 node_access_logs，影响访问明细、IP 汇总与相关趋势查询。',
+                },
+                {
+                  target: 'node_metric_snapshots' as const,
+                  label: '性能快照',
+                  description:
+                    '清理 node_metric_snapshots，影响节点资源趋势和总览资源统计。',
+                },
+                {
+                  target: 'node_request_reports' as const,
+                  label: '请求聚合',
+                  description:
+                    '清理 node_request_reports，影响请求量、错误量与来源聚合展示。',
+                },
+              ].map((item) => (
+                <div
+                  key={item.target}
+                  className="rounded-[28px] border border-[var(--border-default)] bg-[var(--surface-elevated)] p-5"
+                >
+                  <div className="space-y-3">
+                    <div>
+                      <p className="text-lg font-semibold text-[var(--foreground-primary)]">
+                        {item.label}
+                      </p>
+                      <p className="mt-2 text-sm leading-6 text-[var(--foreground-secondary)]">
+                        {item.description}
+                      </p>
+                    </div>
+                    <DangerButton
+                      type="button"
+                      onClick={() => {
+                        setCleanupRetentionDays('');
+                        setCleanupModalState({
+                          target: item.target,
+                          label: item.label,
+                        });
+                      }}
+                    >
+                      清理数据
+                    </DangerButton>
+                  </div>
+                </div>
+              ))}
+            </div>
+          </AppCard>
+        </div>
+      );
+    }
+
    if (activeTab === 'system') {
      return (
        <div className="space-y-6">
@@ -2032,6 +2248,83 @@ export function SettingsPage() {
      </div>

      {renderTabContent()}
+
+      <AppModal
+        isOpen={cleanupModalState !== null}
+        title={`清理${cleanupModalState?.label ?? ''}`}
+        description="输入保留天数后，将只保留该天数范围内的数据；如果留空，则会直接删除该类数据的全部历史记录。"
+        onClose={() => {
+          if (databaseCleanupMutation.isPending) {
+            return;
+          }
+          setCleanupModalState(null);
+        }}
+        footer={
+          <div className="flex flex-wrap justify-end gap-2">
+            <SecondaryButton
+              type="button"
+              onClick={() => setCleanupModalState(null)}
+              disabled={databaseCleanupMutation.isPending}
+            >
+              取消
+            </SecondaryButton>
+            <DangerButton
+              type="button"
+              disabled={databaseCleanupMutation.isPending}
+              onClick={() => {
+                if (!cleanupModalState) {
+                  return;
+                }
+                const trimmed = cleanupRetentionDays.trim();
+                if (trimmed !== '') {
+                  const retentionDays = Number.parseInt(trimmed, 10);
+                  if (Number.isNaN(retentionDays) || retentionDays < 1) {
+                    setFeedback({
+                      tone: 'danger',
+                      message: '手动清理保留天数至少为 1 天。',
+                    });
+                    return;
+                  }
+                  databaseCleanupMutation.mutate({
+                    target: cleanupModalState.target,
+                    retention_days: retentionDays,
+                  });
+                  return;
+                }
+                databaseCleanupMutation.mutate({
+                  target: cleanupModalState.target,
+                });
+              }}
+            >
+              {databaseCleanupMutation.isPending ? '清理中...' : '确认清理'}
+            </DangerButton>
+          </div>
+        }
+      >
+        <div className="space-y-5">
+          <div className="rounded-2xl border border-[var(--status-danger-border)] bg-[var(--status-danger-soft)] px-4 py-4 text-sm leading-6 text-[var(--status-danger-foreground)]">
+            该操作会直接删除数据库中的历史观测数据，删除后无法恢复，请确认当前选择的数据类型和保留范围无误。
+          </div>
+          <ResourceField
+            label="保留天数"
+            hint="留空表示全部删除；填写时必须为大于等于 1 的整数。"
+          >
+            <ResourceInput
+              type="number"
+              min={1}
+              value={cleanupRetentionDays}
+              onChange={(event) => setCleanupRetentionDays(event.target.value)}
+              placeholder="例如 30；留空则全部删除"
+            />
+          </ResourceField>
+          {databaseCleanupMutation.isError ? (
+            <ErrorState
+              title="数据库清理失败"
+              description={getErrorMessage(databaseCleanupMutation.error)}
+            />
+          ) : null}
+        </div>
+      </AppModal>
    </div>
  );
 }
@@ -18,6 +18,25 @@ export interface GeoIPLookupResult {
  longitude?: number | null;
 }

+export type DatabaseCleanupTarget =
+  | 'node_access_logs'
+  | 'node_metric_snapshots'
+  | 'node_request_reports';
+
+export interface DatabaseCleanupPayload {
+  target: DatabaseCleanupTarget;
+  retention_days?: number;
+}
+
+export interface DatabaseCleanupResult {
+  target: DatabaseCleanupTarget;
+  target_label: string;
+  deleted_count: number;
+  delete_all: boolean;
+  retention_days?: number;
+  cutoff?: string;
+}
+
 export interface UpdateSelfPayload {
  username: string;
  display_name: string;
Author	SHA1	Message	Date
ryan	dc72c78b7f	[优化] 界面优化	2026-03-19 21:00:43 +08:00
ryan	9eeccb5fc6	[功能] 添加数据库观测数据清理功能，支持手动和自动清理策略	2026-03-19 20:48:45 +08:00
ryan	a1b3204204	[功能] 添加遗留观察性索引和表的删除逻辑，优化数据库迁移过程	2026-03-19 20:26:18 +08:00
ryan	8737e146d1	[修改] 分片逻辑修改为基于ID	2026-03-19 17:57:30 +08:00
ryan	ae72f2da9a	[功能] 实现数据库版本管理与迁移逻辑，确保数据库结构与版本一致性	2026-03-19 16:45:22 +08:00
ryan	f26fcd028e	[功能] 添加迁移遗留观察性列的功能，支持从 raw_json 填充 metadata_json	2026-03-19 16:31:05 +08:00
ryan	dd49b2777d	[功能] 实现节点访问日志的分片支持，优化日志查询和管理逻辑	2026-03-19 16:19:46 +08:00