mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-08 16:46:37 +08:00
refactor(core): completely decommission legacy internal/apps, internal/platform/bootstrap, and internal/router/v1
This commit is contained in:
@@ -1,240 +0,0 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package bootstrap wires cross-module integrations and process-level subsystem initialization.
|
||||
// All registrations use sync.Once so entry points can call them safely without import-order side effects.
|
||||
package bootstrap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/config"
|
||||
taskhandlers "github.com/Rain-kl/Wavelet/internal/infra/task/handlers"
|
||||
"github.com/Rain-kl/Wavelet/internal/listener"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/platform/lifecycle"
|
||||
"github.com/Rain-kl/Wavelet/internal/repository"
|
||||
"github.com/Rain-kl/Wavelet/internal/repository/logstore"
|
||||
"github.com/Rain-kl/Wavelet/pkg/cache/ram"
|
||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// Options selects role-specific runtime bootstrap steps for the current process.
|
||||
type Options struct {
|
||||
// API enables HTTP-only subsystems such as the ClickHouse access-log writer.
|
||||
API bool
|
||||
}
|
||||
|
||||
// CacheRegistry holds settings for a registered cache type.
|
||||
type CacheRegistry struct {
|
||||
Loader ram.Loader
|
||||
}
|
||||
|
||||
var (
|
||||
registerTasksOnce sync.Once
|
||||
registerPushDomainEventsOnce sync.Once
|
||||
registerTaskListenersOnce sync.Once
|
||||
registerMessageGatewayListenersOnce sync.Once
|
||||
initRuntimeOnce sync.Once
|
||||
|
||||
cacheRegistries = make(map[string]CacheRegistry)
|
||||
cacheRegistriesMu sync.RWMutex
|
||||
|
||||
refreshLocks = make(map[string]*sync.Mutex)
|
||||
refreshLocksMu sync.Mutex
|
||||
)
|
||||
|
||||
// RegisterCache registers a cache type with its Loader for unified preheating and refreshing.
|
||||
func RegisterCache(configType string, reg CacheRegistry) {
|
||||
cacheRegistriesMu.Lock()
|
||||
defer cacheRegistriesMu.Unlock()
|
||||
cacheRegistries[configType] = reg
|
||||
}
|
||||
|
||||
func getRefreshLock(configType string) *sync.Mutex {
|
||||
refreshLocksMu.Lock()
|
||||
defer refreshLocksMu.Unlock()
|
||||
lock, found := refreshLocks[configType]
|
||||
if !found {
|
||||
lock = &sync.Mutex{}
|
||||
refreshLocks[configType] = lock
|
||||
}
|
||||
return lock
|
||||
}
|
||||
|
||||
// PreheatAllCaches preheats all registered caches.
|
||||
func PreheatAllCaches(ctx context.Context) error {
|
||||
cacheRegistriesMu.RLock()
|
||||
defer cacheRegistriesMu.RUnlock()
|
||||
|
||||
for configType, reg := range cacheRegistries {
|
||||
lock := getRefreshLock(configType)
|
||||
lock.Lock()
|
||||
err := ram.Refresh(ctx, configType, "", reg.Loader)
|
||||
lock.Unlock()
|
||||
if err != nil {
|
||||
logger.ErrorF(ctx, "[Bootstrap] preheating cache type %s failed: %v", configType, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RegisterTasks registers all built-in task handlers and metadata.
|
||||
func RegisterTasks() {
|
||||
registerTasksOnce.Do(func() {
|
||||
taskhandlers.Register()
|
||||
})
|
||||
}
|
||||
|
||||
// RegisterPushDomainEvents wires push notification handlers for domain events.
|
||||
func RegisterPushDomainEvents() {
|
||||
registerPushDomainEventsOnce.Do(func() {
|
||||
custom_events.Register()
|
||||
})
|
||||
}
|
||||
|
||||
// RegisterMessageGatewayListeners registers the default log-only inbound handler.
|
||||
func RegisterMessageGatewayListeners() {
|
||||
registerMessageGatewayListenersOnce.Do(func() {
|
||||
listener.OnMessageGatewayInbound(func(ctx context.Context, event listener.MessageGatewayInbound) {
|
||||
userID := uint64(0)
|
||||
if event.Msg.BindingUserID != nil {
|
||||
userID = *event.Msg.BindingUserID
|
||||
}
|
||||
logger.InfoF(ctx, "[%s] channel=%d type=%s user=%d platform_user=%s",
|
||||
listener.EventMessageGatewayInbound,
|
||||
event.Msg.ChannelID,
|
||||
event.Msg.ChannelType,
|
||||
userID,
|
||||
event.Msg.PlatformUserID,
|
||||
)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// RegisterTaskListeners wires operational listeners to task framework hooks.
|
||||
func RegisterTaskListeners() {
|
||||
registerTaskListenersOnce.Do(func() {
|
||||
admin_push.RegisterTaskListeners()
|
||||
})
|
||||
}
|
||||
|
||||
// RegisterAPI wires integrations required by the HTTP API process.
|
||||
func RegisterAPI() {
|
||||
RegisterTasks()
|
||||
RegisterPushDomainEvents()
|
||||
RegisterMessageGatewayListeners()
|
||||
}
|
||||
|
||||
// RegisterWorker wires integrations required by the task worker process.
|
||||
func RegisterWorker() {
|
||||
RegisterTasks()
|
||||
RegisterTaskListeners()
|
||||
RegisterMessageGatewayListeners()
|
||||
}
|
||||
|
||||
// RegisterScheduler wires integrations required by the task scheduler process.
|
||||
func RegisterScheduler() {
|
||||
RegisterTasks()
|
||||
}
|
||||
|
||||
// RegisterAll wires integrations for fused mode (API + Worker + Scheduler).
|
||||
func RegisterAll() {
|
||||
RegisterTasks()
|
||||
RegisterPushDomainEvents()
|
||||
RegisterTaskListeners()
|
||||
RegisterMessageGatewayListeners()
|
||||
}
|
||||
|
||||
// Init runs shared runtime bootstrap exactly once per process.
|
||||
// Call from cmd entry points after wiring registration and database migration, not from router.
|
||||
func Init(ctx context.Context, opts Options) {
|
||||
initRuntimeOnce.Do(func() {
|
||||
if err := validateAndSeedLogDatabase(ctx); err != nil {
|
||||
logger.ErrorF(ctx, "[Bootstrap] 日志主库配置校验失败: %v", err)
|
||||
log.Fatalf("[Bootstrap] 日志主库配置校验失败: %v", err)
|
||||
}
|
||||
|
||||
logstore.SetConfigReader(func(ctx context.Context, key string) (string, error) {
|
||||
cfg, err := repository.GetSystemConfigByKey(ctx, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return cfg.Value, nil
|
||||
})
|
||||
logstore.Init(ctx)
|
||||
|
||||
// Register config cache loader
|
||||
RegisterCache(repository.ConfigCacheType, CacheRegistry{
|
||||
Loader: repository.ConfigLoader{},
|
||||
})
|
||||
|
||||
// Preheat config cache initially (using PreheatAllCaches)
|
||||
if err := PreheatAllCaches(ctx); err != nil {
|
||||
logger.ErrorF(ctx, "[Bootstrap] preheating all caches failed: %v", err)
|
||||
}
|
||||
|
||||
if err := admin_push.SyncEvents(ctx); err != nil {
|
||||
logger.ErrorF(ctx, "[Bootstrap] sync push events failed: %v", err)
|
||||
}
|
||||
if opts.API {
|
||||
risk_control.InitLogWriter(ctx)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func validateAndSeedLogDatabase(ctx context.Context) error {
|
||||
cfg, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyLogDatabase)
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return fmt.Errorf("读取日志主库配置失败: %w", err)
|
||||
}
|
||||
current := cfg.Value
|
||||
if current == "" {
|
||||
current = "sqlite"
|
||||
if config.Config.Database.Enabled {
|
||||
current = "postgres"
|
||||
}
|
||||
if config.Config.ClickHouse.Enabled {
|
||||
current = "clickhouse"
|
||||
}
|
||||
if err := repository.SaveOrUpdateSystemConfig(ctx, model.ConfigKeyLogDatabase, current); err != nil {
|
||||
return fmt.Errorf("初始化日志主库配置失败: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
switch current {
|
||||
case "clickhouse":
|
||||
if !config.Config.ClickHouse.Enabled {
|
||||
return errors.New("当前日志主库为 ClickHouse 但 ClickHouse 未启用。请先重新启用 ClickHouse 配置并启动,在任务管理运行「切换日志数据库」迁移到 PostgreSQL/SQLite 后再禁用 ClickHouse")
|
||||
}
|
||||
case "postgres":
|
||||
if !config.Config.Database.Enabled {
|
||||
return errors.New("当前日志主库为 PostgreSQL 但 PostgreSQL 未启用(当前为 SQLite 主库)。请运行「切换日志数据库」迁回 SQLite 或启用 PostgreSQL")
|
||||
}
|
||||
case "sqlite":
|
||||
if config.Config.Database.Enabled {
|
||||
return errors.New("当前日志主库为 SQLite 但当前主库为 PostgreSQL。请运行「切换日志数据库」迁移到 PostgreSQL")
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("未知的日志主库配置: %s", current)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops all batch writers and background resources.
|
||||
func Stop(ctx context.Context) {
|
||||
lifecycle.Stop(ctx)
|
||||
}
|
||||
|
||||
// ResetInitRuntimeOnceForTest clears initRuntimeOnce so Init can run again in unit tests.
|
||||
func ResetInitRuntimeOnceForTest() {
|
||||
initRuntimeOnce = sync.Once{}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package bootstrap
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/testhelper"
|
||||
)
|
||||
|
||||
func TestInitSyncsPushEventsOnce(t *testing.T) {
|
||||
ResetInitRuntimeOnceForTest()
|
||||
t.Cleanup(ResetInitRuntimeOnceForTest)
|
||||
|
||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
if err := dbConn.AutoMigrate(&model.PushEvent{}); err != nil {
|
||||
t.Fatalf("auto migrate push events failed: %v", err)
|
||||
}
|
||||
|
||||
RegisterPushDomainEvents()
|
||||
|
||||
wantCount := len(admin_push.BuiltInEvents)
|
||||
if wantCount < 1 {
|
||||
t.Fatalf("built-in push events = %d, want at least 1", wantCount)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
Init(ctx, Options{API: true})
|
||||
Init(ctx, Options{}) // second Init must not duplicate events (initRuntimeOnce)
|
||||
|
||||
var count int64
|
||||
if err := dbConn.Model(&model.PushEvent{}).Count(&count).Error; err != nil {
|
||||
t.Fatalf("count push events failed: %v", err)
|
||||
}
|
||||
if count != int64(wantCount) {
|
||||
t.Fatalf("push event count = %d, want %d", count, wantCount)
|
||||
}
|
||||
|
||||
var adminLogin model.PushEvent
|
||||
if err := dbConn.Where("event_key = ?", "admin_login").First(&adminLogin).Error; err != nil {
|
||||
t.Fatalf("admin_login event not found after Init: %v", err)
|
||||
}
|
||||
if adminLogin.Name != "管理员登录" {
|
||||
t.Fatalf("admin_login name = %q, want %q", adminLogin.Name, "管理员登录")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user