refactor(layout): consolidate backend codebase into backend/ package and clean root directory

- Moved cmd/, core/, plugins/, pkg/, downstream/, and main.go into backend/ directory
- Batch updated all Go source files to import github.com/Rain-kl/Wavelet/backend/...
- Updated Makefile, scripts/swagger.sh, architecture guards, and platform skills
- Passed all quality gates (100% tests, 0 lint issues, clean build)
This commit is contained in:
ryan
2026-08-28 12:56:02 +08:00
parent 33b38f8687
commit 43dc97e48c
319 changed files with 912 additions and 1031 deletions
@@ -0,0 +1,119 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package logstore provides data access for analytics tables.
package logstore
import (
"context"
"fmt"
"time"
"github.com/Rain-kl/Wavelet/backend/pkg/util"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
"gorm.io/gorm"
)
// CountAccessLogs returns the number of access logs matching filter.
func CountAccessLogs(ctx context.Context, filter AccessLogFilter) (uint64, error) {
ch := db.ChDB(ctx)
if ch == nil {
return 0, fmt.Errorf("clickhouse gorm connection is not initialized")
}
var count int64
query := applyFilter(ch.Model(&UserAccessLog{}), filter)
if err := query.Count(&count).Error; err != nil {
return 0, fmt.Errorf("count access logs: %w", err)
}
return safeUint64Count(count), nil
}
// ListAccessLogs returns paginated access logs and the total match count.
func ListAccessLogs(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
ch := db.ChDB(ctx)
if ch == nil {
return nil, 0, fmt.Errorf("clickhouse gorm connection is not initialized")
}
if filter.UserIDs != nil && len(filter.UserIDs) == 0 {
return []UserAccessLog{}, 0, nil
}
var total int64
baseQuery := applyFilter(ch.Model(&UserAccessLog{}), filter)
if err := baseQuery.Count(&total).Error; err != nil {
return nil, 0, fmt.Errorf("count access logs: %w", err)
}
if total == 0 {
return []UserAccessLog{}, 0, nil
}
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
offset := (page - 1) * pageSize
var logs []UserAccessLog
err := applyFilter(ch.Model(&UserAccessLog{}), filter).
Order("created_at DESC, id DESC").
Limit(pageSize).
Offset(offset).
Find(&logs).Error
if err != nil {
return nil, 0, fmt.Errorf("list access logs: %w", err)
}
return logs, safeUint64Count(total), nil
}
// DeleteAllUserAccessLogs hard-deletes all user access logs via TRUNCATE.
func DeleteAllUserAccessLogs(ctx context.Context) (int64, error) {
if db.ChConn == nil {
return 0, fmt.Errorf("clickhouse connection is not initialized")
}
if err := db.ChConn.Exec(ctx, "TRUNCATE TABLE "+UserAccessLog{}.TableName()); err != nil {
return 0, fmt.Errorf("truncate user access logs: %w", err)
}
return 0, nil
}
// DeleteUserAccessLogsBefore deletes user access logs older than cutoff.
func DeleteUserAccessLogsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
if db.ChConn == nil {
return 0, fmt.Errorf("clickhouse connection is not initialized")
}
if err := db.ChConn.Exec(ctx, "ALTER TABLE "+UserAccessLog{}.TableName()+" DELETE WHERE created_at < ?", cutoff); err != nil {
return 0, fmt.Errorf("delete expired user access logs: %w", err)
}
return 0, nil
}
func safeUint64Count(count int64) uint64 {
if count < 0 {
return 0
}
return uint64(count)
}
func applyFilter(query *gorm.DB, filter AccessLogFilter) *gorm.DB {
if filter.UserIDs != nil {
if len(filter.UserIDs) == 0 {
return query.Where("1 = 0")
}
query = query.Where("user_id IN ?", filter.UserIDs)
}
if filter.Path != "" {
query = query.Where("path LIKE ?", "%"+util.EscapeLike(filter.Path)+"%")
}
if filter.StartTime != nil {
query = query.Where("created_at >= ?", *filter.StartTime)
}
if filter.EndTime != nil {
query = query.Where("created_at <= ?", *filter.EndTime)
}
return query
}
@@ -0,0 +1,35 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import "time"
// AccessLogFilter scopes user access log queries.
type AccessLogFilter struct {
// UserIDs filters by user IDs. nil means no user filter; an empty slice means no matches.
UserIDs []uint64
Path string
// StartTime filters created_at >= StartTime when non-nil.
StartTime *time.Time
// EndTime filters created_at <= EndTime when non-nil.
EndTime *time.Time
}
// DailyTrend is a single day's access count.
type DailyTrend struct {
Date string
Count uint64
}
// BrowserShare is a browser group's share of access logs.
type BrowserShare struct {
Browser string
Count uint64
}
// TopUser is an active user ranked by access count.
type TopUser struct {
UserID uint64
Count uint64
}
@@ -0,0 +1,140 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"fmt"
"sort"
"time"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
)
const hoursInDay = 24
// GetDailyTrend returns per-day access counts for the last days days (inclusive of today).
func GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
if days < 1 {
days = 7
}
ch := db.ChDB(ctx)
if ch == nil {
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
}
startTime := time.Now().AddDate(0, 0, -(days - 1)).Truncate(hoursInDay * time.Hour)
tableName := UserAccessLog{}.TableName()
query := fmt.Sprintf(`
SELECT toDate(created_at) AS date, count() AS count
FROM %s
WHERE created_at >= ?
GROUP BY date
ORDER BY date ASC
`, tableName)
type trendRow struct {
Date time.Time
Count uint64
}
var rows []trendRow
if err := ch.Raw(query, startTime).Scan(&rows).Error; err != nil {
return nil, fmt.Errorf("get daily trend: %w", err)
}
trendMap := make(map[string]uint64, days)
for i := 0; i < days; i++ {
dateStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
trendMap[dateStr] = 0
}
for _, row := range rows {
dateStr := row.Date.Format("2006-01-02")
trendMap[dateStr] = row.Count
}
result := make([]DailyTrend, 0, days)
for i := days - 1; i >= 0; i-- {
dateStr := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
result = append(result, DailyTrend{
Date: dateStr,
Count: trendMap[dateStr],
})
}
return result, nil
}
// GetBrowserDistribution returns browser-grouped access counts since startTime.
func GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
ch := db.ChDB(ctx)
if ch == nil {
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
}
tableName := UserAccessLog{}.TableName()
query := fmt.Sprintf(`
SELECT user_agent, count() AS count
FROM %s
WHERE created_at >= ?
GROUP BY user_agent
`, tableName)
type uaRow struct {
UserAgent string
Count uint64
}
var rows []uaRow
if err := ch.Raw(query, startTime).Scan(&rows).Error; err != nil {
return nil, fmt.Errorf("get browser distribution: %w", err)
}
browserCounts := make(map[string]uint64)
for _, row := range rows {
browser := ParseBrowserName(row.UserAgent)
browserCounts[browser] += row.Count
}
result := make([]BrowserShare, 0, len(browserCounts))
for browser, count := range browserCounts {
result = append(result, BrowserShare{
Browser: browser,
Count: count,
})
}
sort.Slice(result, func(i, j int) bool {
return result[i].Count > result[j].Count
})
return result, nil
}
// GetTopActiveUsers returns the most active users since startTime.
func GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
if limit < 1 {
limit = 10
}
ch := db.ChDB(ctx)
if ch == nil {
return nil, fmt.Errorf("clickhouse gorm connection is not initialized")
}
tableName := UserAccessLog{}.TableName()
query := fmt.Sprintf(`
SELECT user_id, count() AS count
FROM %s
WHERE created_at >= ? AND user_id > 0
GROUP BY user_id
ORDER BY count DESC
LIMIT ?
`, tableName)
var users []TopUser
if err := ch.Raw(query, startTime, limit).Scan(&users).Error; err != nil {
return nil, fmt.Errorf("get top active users: %w", err)
}
return users, nil
}
@@ -0,0 +1,215 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"io"
"testing"
"time"
"github.com/ClickHouse/clickhouse-go/v2/lib/column"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
)
func setupChGormDB(t *testing.T) *gorm.DB {
t.Helper()
gormDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
DisableForeignKeyConstraintWhenMigrating: true,
})
require.NoError(t, err)
require.NoError(t, gormDB.AutoMigrate(&UserAccessLog{}))
db.SetChDBForTest(gormDB)
return gormDB
}
func TestParseBrowserName(t *testing.T) {
tests := []struct {
name string
ua string
want string
}{
{name: "chrome", ua: "Mozilla/5.0 Chrome/120.0.0.0", want: "Chrome"},
{name: "firefox", ua: "Mozilla/5.0 Firefox/121.0", want: "Firefox"},
{name: "safari", ua: "Mozilla/5.0 Safari/605.1.15", want: "Safari"},
{name: "edge", ua: "Mozilla/5.0 Edg/120.0.0.0", want: "Edge"},
{name: "wechat", ua: "MicroMessenger/8.0", want: "WeChat"},
{name: "postman", ua: "PostmanRuntime/7.36.0", want: "Postman"},
{name: "other", ua: "curl/8.0", want: "Other"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, ParseBrowserName(tt.ua))
})
}
}
func TestCountAccessLogs_EmptyUserIDs(t *testing.T) {
setupChGormDB(t)
t.Cleanup(func() { db.SetChDBForTest(nil) })
count, err := CountAccessLogs(context.Background(), AccessLogFilter{UserIDs: []uint64{}})
require.NoError(t, err)
assert.Equal(t, uint64(0), count)
}
func TestListAccessLogs_EmptyUserIDs(t *testing.T) {
setupChGormDB(t)
t.Cleanup(func() { db.SetChDBForTest(nil) })
logs, total, err := ListAccessLogs(context.Background(), AccessLogFilter{UserIDs: []uint64{}}, 1, 20)
require.NoError(t, err)
assert.Equal(t, uint64(0), total)
assert.Empty(t, logs)
}
func TestListAccessLogs_WithFilters(t *testing.T) {
gormDB := setupChGormDB(t)
t.Cleanup(func() { db.SetChDBForTest(nil) })
now := time.Now().UTC().Truncate(time.Second)
logs := []UserAccessLog{
{ID: 1, UserID: 10, Path: "/api/v1/users", Method: "GET", Status: 200, CreatedAt: now},
{ID: 2, UserID: 20, Path: "/api/v1/admin/logs", Method: "GET", Status: 200, CreatedAt: now},
{ID: 3, UserID: 10, Path: "/api/v1/other", Method: "POST", Status: 201, CreatedAt: now},
}
require.NoError(t, gormDB.Create(&logs).Error)
start := now.Add(-time.Hour)
filter := AccessLogFilter{
UserIDs: []uint64{10},
Path: "users",
StartTime: &start,
}
count, err := CountAccessLogs(context.Background(), filter)
require.NoError(t, err)
assert.Equal(t, uint64(1), count)
result, total, err := ListAccessLogs(context.Background(), filter, 1, 10)
require.NoError(t, err)
assert.Equal(t, uint64(1), total)
require.Len(t, result, 1)
assert.Equal(t, uint64(1), result[0].ID)
assert.Equal(t, "/api/v1/users", result[0].Path)
}
func TestBatchInsert_Empty(t *testing.T) {
err := BatchInsert(context.Background(), nil)
require.NoError(t, err)
}
func TestBatchInsert_UsesModelBatchSQL(t *testing.T) {
ctx := context.Background()
mockBatch := &mockBatch{}
mockConn := &mockConn{
batch: mockBatch,
batchQuery: UserAccessLog{}.BatchInsertSQL(),
}
db.SetChConnForTest(mockConn)
t.Cleanup(func() { db.SetChConnForTest(nil) })
createdAt := time.Now().UTC()
err := BatchInsert(ctx, []UserAccessLog{
{
ID: 1,
UserID: 42,
Path: "/api/v1/test",
Method: "GET",
IP: "127.0.0.1",
UserAgent: "test-agent",
Headers: "{}",
Status: 200,
Latency: 12,
CreatedAt: createdAt,
},
})
require.NoError(t, err)
assert.True(t, mockConn.prepareCalled)
assert.Equal(t, UserAccessLog{}.BatchInsertSQL(), mockConn.preparedQuery)
assert.True(t, mockBatch.sendCalled)
require.Len(t, mockBatch.rows, 1)
assert.Equal(t, uint64(42), mockBatch.rows[0][1])
}
type mockConn struct {
batch driver.Batch
batchQuery string
prepareCalled bool
preparedQuery string
}
func (m *mockConn) Contributors() []string { return nil }
func (m *mockConn) ServerVersion() (*driver.ServerVersion, error) { return nil, nil }
func (m *mockConn) Select(_ context.Context, _ any, _ string, _ ...any) error { return nil }
func (m *mockConn) Query(_ context.Context, _ string, _ ...any) (driver.Rows, error) {
return nil, nil
}
func (m *mockConn) QueryRow(_ context.Context, _ string, _ ...any) driver.Row { return nil }
func (m *mockConn) PrepareBatch(_ context.Context, query string, _ ...driver.PrepareBatchOption) (driver.Batch, error) {
m.prepareCalled = true
m.preparedQuery = query
return m.batch, nil
}
func (m *mockConn) Exec(_ context.Context, _ string, _ ...any) error { return nil }
func (m *mockConn) AsyncInsert(_ context.Context, _ string, _ bool, _ ...any) error { return nil }
func (m *mockConn) InsertFormat(_ context.Context, _ string, _ string, _ io.Reader) error { return nil }
func (m *mockConn) QueryFormat(_ context.Context, _ string, _ string, _ ...any) (io.ReadCloser, error) {
return nil, nil
}
func (m *mockConn) Ping(_ context.Context) error { return nil }
func (m *mockConn) Stats() driver.Stats { return driver.Stats{} }
func (m *mockConn) Close() error { return nil }
type mockBatch struct {
rows [][]any
sendCalled bool
}
func (m *mockBatch) Abort() error { return nil }
func (m *mockBatch) Append(v ...any) error {
m.rows = append(m.rows, v)
return nil
}
func (m *mockBatch) AppendStruct(_ any) error { return nil }
func (m *mockBatch) Column(_ int) driver.BatchColumn { return nil }
func (m *mockBatch) Flush() error { return nil }
func (m *mockBatch) Send() error {
m.sendCalled = true
return nil
}
func (m *mockBatch) IsSent() bool { return m.sendCalled }
func (m *mockBatch) Rows() int { return len(m.rows) }
func (m *mockBatch) Columns() []column.Interface { return nil }
func (m *mockBatch) Close() error { return nil }
@@ -0,0 +1,48 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"fmt"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
)
// BatchInsert writes access logs to ClickHouse using the native batch API.
func BatchInsert(ctx context.Context, logs []UserAccessLog) error {
if len(logs) == 0 {
return nil
}
if db.ChConn == nil {
return fmt.Errorf("clickhouse connection is not initialized")
}
batch, err := db.ChConn.PrepareBatch(ctx, UserAccessLog{}.BatchInsertSQL())
if err != nil {
return fmt.Errorf("prepare clickhouse batch: %w", err)
}
for _, logItem := range logs {
if err := batch.Append(
logItem.ID,
logItem.UserID,
logItem.Path,
logItem.Method,
logItem.IP,
logItem.UserAgent,
logItem.Headers,
logItem.Status,
logItem.Latency,
logItem.CreatedAt,
); err != nil {
return fmt.Errorf("append access log to batch: %w", err)
}
}
if err := batch.Send(); err != nil {
return fmt.Errorf("send clickhouse batch: %w", err)
}
return nil
}
@@ -0,0 +1,30 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import "strings"
// ParseBrowserName performs lightweight User-Agent browser identification.
func ParseBrowserName(ua string) string {
uaLower := strings.ToLower(ua)
if strings.Contains(uaLower, "micromessenger") {
return "WeChat"
}
if strings.Contains(uaLower, "postman") {
return "Postman"
}
if strings.Contains(uaLower, "edg/") || strings.Contains(uaLower, "edge") {
return "Edge"
}
if strings.Contains(uaLower, "firefox") {
return "Firefox"
}
if strings.Contains(uaLower, "chrome") {
return "Chrome"
}
if strings.Contains(uaLower, "safari") {
return "Safari"
}
return "Other"
}
@@ -0,0 +1,99 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"errors"
"fmt"
"strconv"
"time"
"github.com/Rain-kl/Wavelet/backend/pkg/logger"
)
const (
defaultLogRetentionDays = 30
partitionLeadMonths = 2
userAccessLogTable = "w_user_access_logs"
)
// CleanupSummary 汇总本次清理结果。
type CleanupSummary struct {
ActiveDatabase string `json:"active_database"`
RetentionDays int `json:"retention_days"`
Deleted int64 `json:"deleted"`
}
// CleanupExpired 按当前日志库保留天数删除过期用户访问日志,并预建 PG 分区。
func CleanupExpired(ctx context.Context) (CleanupSummary, error) {
active, err := ActiveDatabase(ctx)
if err != nil {
return CleanupSummary{}, err
}
days := retentionDaysForDatabase(ctx, active)
summary := CleanupSummary{ActiveDatabase: active, RetentionDays: days}
store, err := Active(ctx)
if err != nil {
return summary, err
}
now := time.Now().UTC()
if err := store.UserAccessLogs.EnsurePartitions(ctx, now, now.AddDate(0, partitionLeadMonths, 0)); err != nil {
logger.WarnF(ctx, "logstore: ensure partitions during cleanup failed: %v", err)
}
cutoff := now.AddDate(0, 0, -days)
// 先 DROP 完全过期的整月分区,再对边界月逐行 DeleteBefore。
if err := store.UserAccessLogs.DropExpiredPartitions(ctx, cutoff); err != nil {
return summary, fmt.Errorf("drop expired partitions: %w", err)
}
deleted, err := store.UserAccessLogs.DeleteBefore(ctx, cutoff)
if err != nil {
return summary, fmt.Errorf("delete expired user access logs: %w", err)
}
summary.Deleted = deleted
if err := store.UserAccessLogs.DropEmptyPartitions(ctx, now); err != nil {
logger.WarnF(ctx, "drop empty log partitions failed: %v", err)
}
return summary, nil
}
func retentionDaysForDatabase(ctx context.Context, dbName string) int {
key := "log_retention_days_postgres"
switch dbName {
case dbNameSQLite:
key = "log_retention_days_sqlite"
case dbNameClickHouse:
key = "log_retention_days_clickhouse"
}
v, err := getConfig(ctx, key)
if err != nil {
if !errors.Is(err, errConfigReaderNotWired) {
logger.ErrorF(ctx, "读取日志保留天数配置失败(key=%s),回退默认 %d 天: %v", key, defaultLogRetentionDays, err)
}
return defaultLogRetentionDays
}
days, perr := strconv.Atoi(v)
if perr != nil || days <= 0 {
logger.ErrorF(ctx, "日志保留天数配置非法(key=%s, value=%q),回退默认 %d 天", key, v, defaultLogRetentionDays)
return defaultLogRetentionDays
}
return days
}
func partitionStatementsRange(from, to time.Time) []string {
var out []string
start := time.Date(from.Year(), from.Month(), 1, 0, 0, 0, 0, time.UTC)
end := time.Date(to.Year(), to.Month(), 1, 0, 0, 0, 0, time.UTC).AddDate(0, 1, 0)
for ; start.Before(end); start = start.AddDate(0, 1, 0) {
monthEnd := start.AddDate(0, 1, 0)
suffix := start.Format("200601")
fromDay := start.Format("2006-01-02")
toDay := monthEnd.Format("2006-01-02")
out = append(out, fmt.Sprintf(
"CREATE TABLE IF NOT EXISTS %s_%s PARTITION OF %s FOR VALUES FROM ('%s') TO ('%s')",
userAccessLogTable, suffix, userAccessLogTable, fromDay, toDay))
}
return out
}
@@ -0,0 +1,152 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"fmt"
"time"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
)
type clickhouseUserAccessLogStore struct {
skipFreeze bool
}
func newClickHouseUserAccessLogStore() *clickhouseUserAccessLogStore {
return &clickhouseUserAccessLogStore{}
}
var (
_ UserAccessLogStore = (*clickhouseUserAccessLogStore)(nil)
_ StatusStore = (*clickhouseUserAccessLogStore)(nil)
)
func (s *clickhouseUserAccessLogStore) ActiveDatabase(_ context.Context) (string, error) {
return dbNameClickHouse, nil
}
func (s *clickhouseUserAccessLogStore) ensureWritable(ctx context.Context) error {
if !s.skipFreeze && Migrating(ctx) {
return ErrMigrating
}
return nil
}
func (s *clickhouseUserAccessLogStore) BatchInsert(ctx context.Context, logs []UserAccessLog) error {
if len(logs) == 0 {
return nil
}
if err := s.ensureWritable(ctx); err != nil {
return err
}
return BatchInsert(ctx, logs)
}
func (s *clickhouseUserAccessLogStore) DeleteAll(ctx context.Context) (int64, error) {
if err := s.ensureWritable(ctx); err != nil {
return 0, err
}
return DeleteAllUserAccessLogs(ctx)
}
func (s *clickhouseUserAccessLogStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
if err := s.ensureWritable(ctx); err != nil {
return 0, err
}
return DeleteUserAccessLogsBefore(ctx, cutoff)
}
func (s *clickhouseUserAccessLogStore) Count(ctx context.Context, filter AccessLogFilter) (uint64, error) {
return CountAccessLogs(ctx, filter)
}
func (s *clickhouseUserAccessLogStore) List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
return ListAccessLogs(ctx, filter, page, pageSize)
}
func (s *clickhouseUserAccessLogStore) GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
return GetDailyTrend(ctx, days)
}
func (s *clickhouseUserAccessLogStore) GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
return GetBrowserDistribution(ctx, startTime)
}
func (s *clickhouseUserAccessLogStore) GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
return GetTopActiveUsers(ctx, startTime, limit)
}
func (s *clickhouseUserAccessLogStore) EnsurePartitions(_ context.Context, _, _ time.Time) error {
return nil
}
func (s *clickhouseUserAccessLogStore) DropEmptyPartitions(_ context.Context, _ time.Time) error {
return nil
}
func (s *clickhouseUserAccessLogStore) DropExpiredPartitions(_ context.Context, _ time.Time) error {
return nil
}
func (s *clickhouseUserAccessLogStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
if db.ChConn == nil {
return time.Time{}, time.Time{}, fmt.Errorf("clickhouse connection is not initialized")
}
table := UserAccessLog{}.TableName()
var minTime, maxTime *time.Time
if err := db.ChConn.QueryRow(ctx, "SELECT min(created_at), max(created_at) FROM "+table).Scan(&minTime, &maxTime); err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("query migration range %s: %w", table, err)
}
if minTime == nil || maxTime == nil {
return time.Time{}, time.Time{}, nil
}
return minTime.UTC(), maxTime.UTC(), nil
}
func (s *clickhouseUserAccessLogStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error) {
if db.ChConn == nil {
return nil, fmt.Errorf("clickhouse connection is not initialized")
}
if limit <= 0 {
limit = migrationPageSize
}
table := UserAccessLog{}.TableName()
columns := UserAccessLog{}.InsertColumns()
rows, err := db.ChConn.Query(ctx, fmt.Sprintf(
"SELECT %s FROM %s WHERE id > ? ORDER BY id ASC LIMIT ?",
columns, table,
), afterID, limit)
if err != nil {
return nil, fmt.Errorf("list user access logs for migration: %w", err)
}
defer func() { _ = rows.Close() }()
return scanUserAccessLogs(rows)
}
func scanUserAccessLogs(rows driver.Rows) ([]UserAccessLog, error) {
var result []UserAccessLog
for rows.Next() {
var item UserAccessLog
if err := rows.Scan(
&item.ID,
&item.UserID,
&item.Path,
&item.Method,
&item.IP,
&item.UserAgent,
&item.Headers,
&item.Status,
&item.Latency,
&item.CreatedAt,
); err != nil {
return nil, fmt.Errorf("scan user access log row: %w", err)
}
item.CreatedAt = item.CreatedAt.UTC()
result = append(result, item)
}
return result, nil
}
@@ -0,0 +1,325 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"errors"
"fmt"
"sort"
"strings"
"time"
"github.com/Rain-kl/Wavelet/backend/pkg/idgen"
"gorm.io/gorm"
)
const (
insertBatchSize = 500
migrationPageSize = 100
defaultPageSize = 20
defaultTopN = 10
topUserAgents = 100
dayDuration = 24 * time.Hour
)
type gormLogStore struct {
db *gorm.DB
skipFreeze bool
}
func newGormStore(db *gorm.DB) *gormLogStore { return &gormLogStore{db: db} }
type userAccessLogGormStore struct {
*gormLogStore
}
func newUserAccessLogGormStore(db *gorm.DB) *userAccessLogGormStore {
return &userAccessLogGormStore{gormLogStore: newGormStore(db)}
}
var (
_ UserAccessLogStore = (*userAccessLogGormStore)(nil)
_ StatusStore = (*userAccessLogGormStore)(nil)
)
func (s *gormLogStore) ActiveDatabase(_ context.Context) (string, error) {
if isPostgresDialect(s.db) {
return dbNamePostgres, nil
}
return dbNameSQLite, nil
}
func (s *gormLogStore) ensureWritable(ctx context.Context) error {
if !s.skipFreeze && Migrating(ctx) {
return ErrMigrating
}
return nil
}
func (s *userAccessLogGormStore) BatchInsert(ctx context.Context, logs []UserAccessLog) error {
if len(logs) == 0 {
return nil
}
if err := s.ensureWritable(ctx); err != nil {
return err
}
for i := range logs {
if logs[i].ID == 0 {
logs[i].ID = idgen.NextUint64ID()
}
}
return s.db.WithContext(ctx).CreateInBatches(logs, insertBatchSize).Error
}
func (s *userAccessLogGormStore) DeleteAll(ctx context.Context) (int64, error) {
if err := s.ensureWritable(ctx); err != nil {
return 0, err
}
res := s.db.WithContext(ctx).Where("1 = 1").Delete(&UserAccessLog{})
return res.RowsAffected, res.Error
}
func (s *userAccessLogGormStore) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) {
if err := s.ensureWritable(ctx); err != nil {
return 0, err
}
res := s.db.WithContext(ctx).Where("created_at < ?", cutoff).Delete(&UserAccessLog{})
if res.Error != nil && isMissingRelation(res.Error) {
return 0, nil
}
return res.RowsAffected, res.Error
}
func (s *userAccessLogGormStore) ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error) {
var rows []UserAccessLog
q := s.db.WithContext(ctx).Model(&UserAccessLog{}).
Where("id > ?", afterID).
Order("id ASC").
Limit(limitOr(limit, migrationPageSize))
if err := q.Find(&rows).Error; err != nil {
return nil, err
}
return rows, nil
}
func (s *userAccessLogGormStore) MigrationRange(ctx context.Context) (time.Time, time.Time, error) {
return gormMigrationRange(ctx, s.db, "created_at", UserAccessLog{}, func(v *UserAccessLog) time.Time {
return v.CreatedAt
})
}
func (s *userAccessLogGormStore) Count(ctx context.Context, filter AccessLogFilter) (uint64, error) {
where, args, ok := buildUserAccessLogWhere(filter)
if !ok {
return 0, nil
}
var total int64
if err := s.db.WithContext(ctx).Model(&UserAccessLog{}).Where(where, args...).Count(&total).Error; err != nil {
return 0, err
}
return countToUint64(total), nil
}
func (s *userAccessLogGormStore) List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error) {
where, args, ok := buildUserAccessLogWhere(filter)
if !ok {
return []UserAccessLog{}, 0, nil
}
var total int64
if err := s.db.WithContext(ctx).Model(&UserAccessLog{}).Where(where, args...).Count(&total).Error; err != nil {
return nil, 0, err
}
if total == 0 {
return []UserAccessLog{}, 0, nil
}
var rows []UserAccessLog
q := s.db.WithContext(ctx).Where(where, args...).Order("created_at DESC, id DESC")
if err := q.Limit(limitOr(pageSize, defaultPageSize)).Offset(offsetOf(page, pageSize)).Find(&rows).Error; err != nil {
return nil, 0, err
}
return rows, countToUint64(total), nil
}
func buildUserAccessLogWhere(filter AccessLogFilter) (string, []any, bool) {
if filter.UserIDs != nil && len(filter.UserIDs) == 0 {
return "", nil, false
}
var parts []string
var args []any
if filter.UserIDs != nil {
parts = append(parts, "user_id IN ?")
args = append(args, filter.UserIDs)
}
if trimmed := strings.TrimSpace(filter.Path); trimmed != "" {
parts = append(parts, "path LIKE ?")
args = append(args, "%"+trimmed+"%")
}
if filter.StartTime != nil {
parts = append(parts, "created_at >= ?")
args = append(args, *filter.StartTime)
}
if filter.EndTime != nil {
parts = append(parts, "created_at <= ?")
args = append(args, *filter.EndTime)
}
if len(parts) == 0 {
return "1 = 1", args, true
}
return strings.Join(parts, " AND "), args, true
}
func (s *userAccessLogGormStore) GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error) {
if days <= 0 {
days = 7
}
start := time.Now().AddDate(0, 0, -(days - 1)).Truncate(dayDuration)
type row struct {
Date string
Cnt uint64
}
var rows []row
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
Select(dailyTrendDateSQL(s.db)+" AS date, COUNT(*) AS cnt").
Where("created_at >= ?", start).
Group("date").Order("date ASC").Scan(&rows).Error
if err != nil {
return nil, err
}
counts := make(map[string]uint64, len(rows))
for _, r := range rows {
counts[r.Date] = r.Cnt
}
out := make([]DailyTrend, 0, days)
for i := 0; i < days; i++ {
d := start.AddDate(0, 0, i).Format("2006-01-02")
out = append(out, DailyTrend{Date: d, Count: counts[d]})
}
return out, nil
}
func (s *userAccessLogGormStore) GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error) {
type row struct {
UserAgent string
Cnt uint64
}
var rows []row
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
Select("user_agent, COUNT(*) AS cnt").
Where("created_at >= ?", startTime).
Group("user_agent").Order("cnt DESC").Limit(topUserAgents).Scan(&rows).Error
if err != nil {
return nil, err
}
counts := make(map[string]uint64)
for _, r := range rows {
counts[ParseBrowserName(r.UserAgent)] += r.Cnt
}
out := make([]BrowserShare, 0, len(counts))
for label, count := range counts {
out = append(out, BrowserShare{Browser: label, Count: count})
}
sort.Slice(out, func(i, j int) bool { return out[i].Count > out[j].Count })
return out, nil
}
func (s *userAccessLogGormStore) GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error) {
type row struct {
UserID uint64
Cnt uint64
}
var rows []row
err := s.db.WithContext(ctx).Model(&UserAccessLog{}).
Select("user_id, COUNT(*) AS cnt").
Where("user_id <> 0 AND created_at >= ?", startTime).
Group("user_id").Order("cnt DESC").Limit(limitOr(limit, defaultTopN)).Scan(&rows).Error
if err != nil {
return nil, err
}
out := make([]TopUser, len(rows))
for i, r := range rows {
out[i] = TopUser{UserID: r.UserID, Count: r.Cnt}
}
return out, nil
}
func (s *userAccessLogGormStore) EnsurePartitions(ctx context.Context, from, to time.Time) error {
if !isPostgresDialect(s.db) {
return nil
}
for _, sql := range partitionStatementsRange(from, to) {
if err := s.db.WithContext(ctx).Exec(sql).Error; err != nil {
return fmt.Errorf("ensure partition: %w", err)
}
}
return nil
}
func gormMigrationRange[T any](
ctx context.Context,
gdb *gorm.DB,
column string,
model T,
timeOf func(*T) time.Time,
) (time.Time, time.Time, error) {
var first, last T
found := false
for _, order := range []string{"ASC", "DESC"} {
out := &first
if order == "DESC" {
out = &last
}
res := gdb.WithContext(ctx).Model(model).Order(column + " " + order).Limit(1).Take(out)
if res.Error != nil && !errors.Is(res.Error, gorm.ErrRecordNotFound) {
return time.Time{}, time.Time{}, fmt.Errorf("query migration range %s: %w", column, res.Error)
}
if res.Error == nil {
found = true
}
}
if !found {
return time.Time{}, time.Time{}, nil
}
return timeOf(&first).UTC(), timeOf(&last).UTC(), nil
}
func limitOr(v, def int) int {
if v <= 0 {
return def
}
return v
}
func offsetOf(page, pageSize int) int {
if page < 1 {
page = 1
}
return (page - 1) * limitOr(pageSize, defaultPageSize)
}
func countToUint64(v int64) uint64 {
if v < 0 {
return 0
}
return uint64(v)
}
func isPostgresDialect(db *gorm.DB) bool {
return db != nil && db.Dialector != nil && db.Name() == "postgres"
}
func dailyTrendDateSQL(db *gorm.DB) string {
if isPostgresDialect(db) {
return "to_char(created_at, 'YYYY-MM-DD')"
}
return "strftime('%Y-%m-%d', created_at)"
}
func isMissingRelation(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "no such table") || strings.Contains(msg, "does not exist")
}
@@ -0,0 +1,58 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"testing"
"time"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
)
func newTestUserAccessStore(t *testing.T) *userAccessLogGormStore {
t.Helper()
gdb, err := gorm.Open(sqlite.Open("file:logstore-"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
require.NoError(t, err)
require.NoError(t, gdb.AutoMigrate(&UserAccessLog{}))
return newUserAccessLogGormStore(gdb)
}
func TestGormUserAccessLogCountList(t *testing.T) {
ua := newTestUserAccessStore(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Second)
require.NoError(t, ua.BatchInsert(ctx, []UserAccessLog{
{UserID: 10, Path: "/api/v1/users", Method: "GET", Status: 200, CreatedAt: now},
{UserID: 20, Path: "/api/v1/admin", Method: "GET", Status: 200, CreatedAt: now},
{UserID: 10, Path: "/api/v1/other", Method: "POST", Status: 201, CreatedAt: now},
}))
count, err := ua.Count(ctx, AccessLogFilter{UserIDs: []uint64{10}, Path: "users"})
require.NoError(t, err)
require.Equal(t, uint64(1), count)
rows, total, err := ua.List(ctx, AccessLogFilter{UserIDs: []uint64{10}, Path: "users"}, 1, 10)
require.NoError(t, err)
require.Equal(t, uint64(1), total)
require.Len(t, rows, 1)
require.Equal(t, "/api/v1/users", rows[0].Path)
require.NotZero(t, rows[0].ID)
}
func TestGormUserAccessLogFreeze(t *testing.T) {
ua := newTestUserAccessStore(t)
SetConfigReader(func(_ context.Context, key string) (string, error) {
if key == logMigrationKey {
return "migrating", nil
}
return "", nil
})
t.Cleanup(ResetForTest)
err := ua.BatchInsert(context.Background(), []UserAccessLog{{UserID: 1, CreatedAt: time.Now()}})
require.ErrorIs(t, err, ErrMigrating)
}
@@ -0,0 +1,41 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"os/exec"
"strings"
"testing"
)
// plugins/domain 不得直接 import 已废弃的 internal。
var forbiddenImports = []string{
"github.com/Rain-kl/Wavelet/internal",
"github.com/Rain-kl/Wavelet/backend/internal",
}
func TestAppsMustNotImportPersistenceDirectly(t *testing.T) {
t.Chdir("../../../../..")
out, err := exec.Command("go", "list", "-test", "-f", `{{.ImportPath}} {{join .Imports " "}}`, "./backend/plugins/domain/...").Output()
if err != nil {
t.Fatalf("go list: %v", err)
}
for _, line := range strings.Split(string(out), "\n") {
fields := strings.Fields(line)
if len(fields) == 0 {
continue
}
pkg := fields[0]
if !strings.HasPrefix(pkg, "github.com/Rain-kl/Wavelet/backend/plugins/domain") {
continue
}
for _, imp := range fields[1:] {
for _, forbidden := range forbiddenImports {
if imp == forbidden {
t.Errorf("%s must not import forbidden package %s", pkg, forbidden)
}
}
}
}
}
@@ -0,0 +1,46 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package logstore abstracts user access-log storage across ClickHouse, PostgreSQL and SQLite.
package logstore
import (
"context"
"errors"
"time"
)
// ErrMigrating 表示日志数据库正在迁移,当前禁止写入。
var ErrMigrating = errors.New("log database is migrating, writes are disabled")
// UserAccessLogStore 用户访问日志(w_user_access_logs)。
type UserAccessLogStore interface {
BatchInsert(ctx context.Context, logs []UserAccessLog) error
DeleteAll(ctx context.Context) (int64, error)
DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error)
Count(ctx context.Context, filter AccessLogFilter) (uint64, error)
List(ctx context.Context, filter AccessLogFilter, page, pageSize int) ([]UserAccessLog, uint64, error)
GetDailyTrend(ctx context.Context, days int) ([]DailyTrend, error)
GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]BrowserShare, error)
GetTopActiveUsers(ctx context.Context, startTime time.Time, limit int) ([]TopUser, error)
ListForMigration(ctx context.Context, afterID uint64, limit int) ([]UserAccessLog, error)
MigrationRange(ctx context.Context) (from, to time.Time, err error)
EnsurePartitions(ctx context.Context, from, to time.Time) error
// DropEmptyPartitions 幂等清理 PG 空分区表:删除 before 月份之前、且无任何数据的按月分区;
// CH/SQLite 为 no-op。
DropEmptyPartitions(ctx context.Context, before time.Time) error
// DropExpiredPartitions 直接删除完全过期的 PG 整月分区(候选为月份早于 cutoff 月的分区,
// 删除前校验分区内无保留期内数据,避免时区偏移下误删;迁移冻结期间拒绝执行);CH/SQLite 为 no-op。
DropExpiredPartitions(ctx context.Context, cutoff time.Time) error
}
// StatusStore 日志库状态。
type StatusStore interface {
ActiveDatabase(ctx context.Context) (string, error)
}
// Store 当前生效日志库。
type Store struct {
UserAccessLogs UserAccessLogStore
Status StatusStore
}
@@ -0,0 +1,21 @@
-- +goose Up
CREATE TABLE IF NOT EXISTS w_user_access_logs
(
id UInt64,
user_id UInt64,
path String,
method String,
ip String,
user_agent String,
headers String,
status Int32,
latency Int64,
created_at DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (created_at, ip, user_id)
SETTINGS index_granularity = 8192;
-- +goose Down
DROP TABLE IF EXISTS w_user_access_logs;
@@ -0,0 +1,20 @@
-- +goose Up
CREATE TABLE IF NOT EXISTS w_user_access_logs (
id BIGINT NOT NULL,
user_id BIGINT NOT NULL DEFAULT 0,
path VARCHAR(2048) NOT NULL DEFAULT '',
method VARCHAR(16) NOT NULL DEFAULT '',
ip VARCHAR(128) NOT NULL DEFAULT '',
user_agent TEXT NOT NULL DEFAULT '',
headers TEXT NOT NULL DEFAULT '',
status INTEGER NOT NULL DEFAULT 0,
latency BIGINT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_user_id ON w_user_access_logs (user_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_w_user_access_logs_created_at ON w_user_access_logs (created_at DESC);
-- +goose Down
DROP TABLE IF EXISTS w_user_access_logs;
@@ -0,0 +1,44 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package logstore abstracts user access-log storage across ClickHouse, PostgreSQL and SQLite.
package logstore
import (
"fmt"
"time"
)
const (
userAccessLogTableName = "w_user_access_logs"
userAccessLogInsertColumns = "id, user_id, path, method, ip, user_agent, headers, status, latency, created_at"
)
// UserAccessLog stores HTTP access records in ClickHouse, PostgreSQL and SQLite.
type UserAccessLog struct {
ID uint64 `gorm:"column:id"`
UserID uint64 `gorm:"column:user_id"`
Path string `gorm:"column:path"`
Method string `gorm:"column:method"`
IP string `gorm:"column:ip"`
UserAgent string `gorm:"column:user_agent"`
Headers string `gorm:"column:headers"`
Status int32 `gorm:"column:status"`
Latency int64 `gorm:"column:latency"`
CreatedAt time.Time `gorm:"column:created_at"`
}
// TableName returns the table name.
func (UserAccessLog) TableName() string {
return userAccessLogTableName
}
// InsertColumns returns comma-separated column names for batch insert.
func (UserAccessLog) InsertColumns() string {
return userAccessLogInsertColumns
}
// BatchInsertSQL returns the INSERT prefix used by native batch writers.
func (UserAccessLog) BatchInsertSQL() string {
return fmt.Sprintf("INSERT INTO %s (%s)", userAccessLogTableName, userAccessLogInsertColumns)
}
@@ -0,0 +1,115 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"fmt"
"strings"
"time"
"gorm.io/gorm"
)
// listPartitionNames 列出 table 在当前 schema 下的全部直接分区表名(pg_inherits)。
func listPartitionNames(ctx context.Context, gdb *gorm.DB, table string) ([]string, error) {
var names []string
if err := gdb.WithContext(ctx).Raw(`
SELECT c.relname
FROM pg_inherits i
JOIN pg_class c ON c.oid = i.inhrelid
JOIN pg_class p ON p.oid = i.inhparent
JOIN pg_namespace n ON n.oid = p.relnamespace AND n.nspname = current_schema()
WHERE p.relname = ?`, table).Scan(&names).Error; err != nil {
return nil, fmt.Errorf("list partitions of %s: %w", table, err)
}
return names, nil
}
// partitionNameMonth 解析按月分区表名 <table>_YYYYMM 的所属月份;命名不匹配返回 (零值, false)。
func partitionNameMonth(table, name string) (time.Time, bool) {
suffix, ok := strings.CutPrefix(name, table+"_")
if !ok || len(suffix) != 6 {
return time.Time{}, false
}
m, err := time.Parse("200601", suffix)
if err != nil {
return time.Time{}, false
}
return m, true
}
// dropEligiblePartitionNames 返回 before 月份之前、命名合法的分区表名(是否为空由调用方校验)。
func dropEligiblePartitionNames(table string, names []string, before time.Time) []string {
beforeMonth := time.Date(before.Year(), before.Month(), 1, 0, 0, 0, 0, time.UTC)
out := make([]string, 0, len(names))
for _, name := range names {
month, ok := partitionNameMonth(table, name)
if !ok || !month.Before(beforeMonth) {
continue
}
out = append(out, name)
}
return out
}
// DropEmptyPartitions 幂等清理 PG 空分区表:仅删除 before 月份之前、且无任何数据的分区。
// 非 PG 方言为 no-op。
func (s *gormLogStore) DropEmptyPartitions(ctx context.Context, before time.Time) error {
if !isPostgresDialect(s.db) {
return nil
}
names, err := listPartitionNames(ctx, s.db, userAccessLogTable)
if err != nil {
return err
}
for _, name := range dropEligiblePartitionNames(userAccessLogTable, names, before) {
var one int
if err := s.db.WithContext(ctx).Raw("SELECT 1 FROM " + name + " LIMIT 1").Scan(&one).Error; err != nil {
return fmt.Errorf("check partition %s empty: %w", name, err)
}
if one == 1 {
continue
}
if err := s.db.WithContext(ctx).Exec("DROP TABLE IF EXISTS " + name).Error; err != nil {
return fmt.Errorf("drop empty partition %s: %w", name, err)
}
}
return nil
}
// DropExpiredPartitions 直接删除完全过期的 PG 整月分区(避免 retention 清理逐行 DELETE)。
// 候选 = 月份早于 cutoff 月(按 cutoff 的 UTC 时刻取月);删除前校验分区内不存在 created_at >= cutoff 的行。
// 迁移冻结期间返回 ErrMigrating。CH/SQLite 为 no-op。
func (s *gormLogStore) DropExpiredPartitions(ctx context.Context, cutoff time.Time) error {
if !isPostgresDialect(s.db) {
return nil
}
if err := s.ensureWritable(ctx); err != nil {
return err
}
names, err := listPartitionNames(ctx, s.db, userAccessLogTable)
if err != nil {
return err
}
cu := cutoff.UTC()
cutoffMonth := time.Date(cu.Year(), cu.Month(), 1, 0, 0, 0, 0, time.UTC)
for _, name := range names {
month, ok := partitionNameMonth(userAccessLogTable, name)
if !ok || !month.Before(cutoffMonth) {
continue
}
var hasRetained int
if err := s.db.WithContext(ctx).Raw("SELECT 1 FROM "+name+" WHERE created_at >= ? LIMIT 1", cu).Scan(&hasRetained).Error; err != nil {
return fmt.Errorf("check partition %s retained rows: %w", name, err)
}
if hasRetained == 1 {
continue
}
if err := s.db.WithContext(ctx).Exec("DROP TABLE IF EXISTS " + name).Error; err != nil {
return fmt.Errorf("drop expired partition %s: %w", name, err)
}
}
return nil
}
@@ -0,0 +1,69 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestPartitionNameMonth(t *testing.T) {
cases := []struct {
table string
name string
want string
}{
{"w_user_access_logs", "w_user_access_logs_202612", "2026-12"},
{"w_user_access_logs", "w_user_access_logs_202608", "2026-08"},
{"w_user_access_logs", "of_node_access_logs_202608", ""},
{"w_user_access_logs", "w_user_access_logs_20268", ""},
{"w_user_access_logs", "w_user_access_logs_202613", ""},
{"w_user_access_logs", "w_user_access_logs_default", ""},
}
for _, c := range cases {
got, ok := partitionNameMonth(c.table, c.name)
if c.want == "" {
if ok {
t.Fatalf("partitionNameMonth(%q, %q) ok = true, want false", c.table, c.name)
}
continue
}
if !ok || got.Format("2006-01") != c.want {
t.Fatalf("partitionNameMonth(%q, %q) = %v, want %s", c.table, c.name, got, c.want)
}
}
}
func TestDropEligiblePartitionNames(t *testing.T) {
before := time.Date(2026, 10, 15, 0, 0, 0, 0, time.UTC)
names := []string{
"w_user_access_logs_202608",
"w_user_access_logs_202609",
"w_user_access_logs_202610",
"w_user_access_logs_202611",
"w_user_access_logs_default",
}
got := dropEligiblePartitionNames(userAccessLogTable, names, before)
want := []string{"w_user_access_logs_202608", "w_user_access_logs_202609"}
require.Equal(t, want, got)
first := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
require.Empty(t, dropEligiblePartitionNames(userAccessLogTable, []string{"w_user_access_logs_202610"}, first))
}
func TestDropPartitionHelpersSQLiteNoop(t *testing.T) {
ua := newTestUserAccessStore(t)
ctx := context.Background()
require.NoError(t, ua.BatchInsert(ctx, []UserAccessLog{
{UserID: 1, Path: "/x", CreatedAt: time.Now().UTC()},
}))
require.NoError(t, ua.DropExpiredPartitions(ctx, time.Now().AddDate(0, 0, -90)))
require.NoError(t, ua.DropEmptyPartitions(ctx, time.Now()))
count, err := ua.Count(ctx, AccessLogFilter{})
require.NoError(t, err)
require.Equal(t, uint64(1), count)
}
@@ -0,0 +1,188 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package logstore
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/Rain-kl/Wavelet/backend/pkg/config"
"github.com/Rain-kl/Wavelet/backend/pkg/logger"
db "github.com/Rain-kl/Wavelet/backend/plugins/infra/database"
)
const (
logDatabaseKey = "log_database"
logMigrationKey = "log_db_migration"
)
const (
dbNamePostgres = "postgres"
dbNameSQLite = "sqlite"
dbNameClickHouse = "clickhouse"
)
var errConfigReaderNotWired = errors.New("logstore: config reader not wired")
// ConfigReader 读取系统配置字符串值,由 bootstrap 注入(避免 logstore ↔ repository 循环依赖)。
type ConfigReader func(ctx context.Context, key string) (string, error)
const resolveCacheTTL = 1 * time.Second
var (
configReader ConfigReader
storeMu sync.RWMutex
active *Store
activeDB string
lastResolveDB string
lastResolveTime time.Time
)
// SetConfigReader 注入系统配置读取函数(bootstrap 调用,测试可注入内存实现)。
func SetConfigReader(fn ConfigReader) { configReader = fn }
func getConfig(ctx context.Context, key string) (string, error) {
if configReader == nil {
return "", errConfigReaderNotWired
}
return configReader(ctx, key)
}
// Active 返回当前生效的日志库 Store。
func Active(ctx context.Context) (*Store, error) {
current, err := resolveDatabase(ctx)
if err != nil {
return nil, err
}
storeMu.RLock()
if active != nil && activeDB == current {
s := active
storeMu.RUnlock()
return s, nil
}
storeMu.RUnlock()
storeMu.Lock()
defer storeMu.Unlock()
if active != nil && activeDB == current {
return active, nil
}
s, err := buildStore(ctx, current, false)
if err != nil {
return nil, err
}
active = s
activeDB = current
return s, nil
}
// Build 直接按目标构造 store(不经 Active 缓存)。
func Build(ctx context.Context, database string) (*Store, error) {
return buildStore(ctx, database, false)
}
// BuildForMigration 构造迁移目标 store,跳过冻结检查。
func BuildForMigration(ctx context.Context, database string) (*Store, error) {
return buildStore(ctx, database, true)
}
func buildStore(ctx context.Context, database string, skipFreeze bool) (*Store, error) {
switch database {
case dbNameClickHouse:
ual := newClickHouseUserAccessLogStore()
ual.skipFreeze = skipFreeze
return &Store{UserAccessLogs: ual, Status: ual}, nil
case dbNamePostgres, dbNameSQLite:
gdb := db.DB(ctx)
ual := newUserAccessLogGormStore(gdb)
ual.skipFreeze = skipFreeze
return &Store{UserAccessLogs: ual, Status: ual}, nil
default:
return nil, fmt.Errorf("unsupported log database: %s", database)
}
}
// Migrating 返回日志库是否处于迁移冻结状态。
func Migrating(ctx context.Context) bool {
v, err := getConfig(ctx, logMigrationKey)
if err != nil {
if !errors.Is(err, errConfigReaderNotWired) {
logger.ErrorF(ctx, "read log migration config failed: %v", err)
}
return false
}
return v == "migrating"
}
// Init 预热激活 store,并兜底预建当前月及未来分区。
func Init(ctx context.Context) {
s, err := Active(ctx)
if err != nil {
return
}
now := time.Now().UTC()
if err := s.UserAccessLogs.EnsurePartitions(ctx, now, now.AddDate(0, partitionLeadMonths, 0)); err != nil {
logger.WarnF(ctx, "logstore: ensure startup partitions failed: %v", err)
}
}
// InvalidateCache 清空日志库解析缓存。
func InvalidateCache() {
storeMu.Lock()
defer storeMu.Unlock()
lastResolveTime = time.Time{}
lastResolveDB = ""
}
// ResetForTest 清空缓存的激活 store 与 config reader。
func ResetForTest() {
storeMu.Lock()
active = nil
activeDB = ""
lastResolveDB = ""
lastResolveTime = time.Time{}
storeMu.Unlock()
configReader = nil
}
// ActiveDatabase 返回当前日志主库名。
func ActiveDatabase(ctx context.Context) (string, error) {
return resolveDatabase(ctx)
}
func resolveDatabase(ctx context.Context) (string, error) {
storeMu.RLock()
if active != nil && time.Since(lastResolveTime) < resolveCacheTTL {
name := lastResolveDB
storeMu.RUnlock()
return name, nil
}
storeMu.RUnlock()
v, err := getConfig(ctx, logDatabaseKey)
if err != nil && !errors.Is(err, errConfigReaderNotWired) {
return "", err
}
resolved := v
if resolved == "" {
resolved = dbNameSQLite
if config.Config.Database.Enabled {
resolved = dbNamePostgres
}
if config.Config.ClickHouse.Enabled {
resolved = dbNameClickHouse
}
}
storeMu.Lock()
lastResolveDB = resolved
lastResolveTime = time.Now()
storeMu.Unlock()
return resolved, nil
}