diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index 34c5e36..eacb76b 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -9,16 +9,18 @@ package main import ( "github.com/go-kratos/kratos/v3" "kra/internal/app" - integration2 "kra/internal/biz/integration" + integration3 "kra/internal/biz/integration" payment2 "kra/internal/biz/payment" system2 "kra/internal/biz/system" - "kra/internal/biz/task" + task2 "kra/internal/biz/task" "kra/internal/conf" "kra/internal/data" + "kra/internal/data/integration" "kra/internal/data/payment" - "kra/internal/data/repository" + "kra/internal/data/system" + "kra/internal/data/task" "kra/internal/initialize" - "kra/internal/integration" + integration2 "kra/internal/integration" "kra/internal/integration/cache" "kra/internal/integration/email" "kra/internal/integration/mq" @@ -103,15 +105,15 @@ func wireApp(confServer *conf.Server, runtime *conf.Runtime, logger *slog.Logger paymentUsecase := payment2.NewPaymentUsecase(paymentRepo, paymentOrderRepo, logger) paymentService := service.NewPaymentService(paymentUsecase) handlerPayment := handler.NewPayment(paymentService) - taskRepo := system.NewTaskRepo(dataData) + taskRepo := task.NewTaskRepo(dataData) registry := app.TaskRegistry(catalog) - taskUsecase := task.NewTaskUsecaseWithRegistry(taskRepo, registry) + taskUsecase := task2.NewTaskUsecaseWithRegistry(taskRepo, registry) mediaRepo := system.NewMediaRepo(dataData) mediaUsecase := system2.NewMediaUsecase(mediaRepo, reloadable, runtimeSettings) taskExecutor := worker.NewTaskExecutorWithRegistry(taskUsecase, mediaUsecase, runtime, registry) taskScheduler := worker.NewTaskScheduler(taskUsecase, authorityUsecase, taskExecutor, logger) taskRuntime := worker.NewTaskRuntime(taskScheduler) - taskApplicationUsecase := task.NewTaskApplicationUsecase(taskUsecase, taskRuntime) + taskApplicationUsecase := task2.NewTaskApplicationUsecase(taskUsecase, taskRuntime) taskService := service.NewTaskService(taskApplicationUsecase) handlerTask := handler.NewTask(taskService) mediaService := service.NewMediaService(mediaUsecase, runtimeSettings) @@ -151,15 +153,17 @@ func wireApp(confServer *conf.Server, runtime *conf.Runtime, logger *slog.Logger user := handler.NewUser(userService, authService) navigation := handler.NewNavigation(userService) session := handler.NewSession(tokenService) - integrationConfigRepo := system.NewIntegrationConfigRepo(dataData) + integrationConfigRepo := integration.NewIntegrationConfigRepo(dataData) store := data.NewIntegrationRuntime(dataData) - connectivityTester := integration.NewConnectivityTester(store) - integrationConfigUsecase := integration2.NewIntegrationConfigUsecase(integrationConfigRepo, connectivityTester) + connectivityTester := integration2.NewConnectivityTester(store) + integrationConfigUsecase := integration3.NewIntegrationConfigUsecase(integrationConfigRepo, connectivityTester) integrationConfigService := service.NewIntegrationConfigService(integrationConfigUsecase) integrationConfig := handler.NewIntegrationConfig(integrationConfigService) v := handler.NewSet(authority, menu, api, permission, organization, announcement, handlerEmail, handlerPayment, handlerTask, media, audit, export, version, dictionary, parameter, apiToken, systemConfig, public, user, navigation, session, integrationConfig) routes := router.NewRoutes(v) - taskMethods := worker.NewTaskMethods(taskUsecase, mediaUsecase, runtime) + maintenanceRepo := system.NewMaintenanceRepo(dataData) + maintenanceUsecase := system2.NewMaintenanceUsecase(maintenanceRepo) + taskMethods := worker.NewTaskMethods(taskUsecase, maintenanceUsecase, mediaUsecase, runtime) appRuntimeContributions := runtimeContributions(routes, taskMethods) moduleRuntime := app.Runtime(appRuntimeContributions, registry) websocketServer, cleanup2, err := websocket.New(store) diff --git a/docs/OA_MIGRATION_PLAN.md b/docs/OA_MIGRATION_PLAN.md index cdb249f..772c114 100644 --- a/docs/OA_MIGRATION_PLAN.md +++ b/docs/OA_MIGRATION_PLAN.md @@ -85,7 +85,7 @@ web/src/modules// components/ ``` -修改 KRA 既有系统表和系统行为的内容仍放在现有 `internal/biz`、`internal/data/repository`、`internal/service`、`internal/server` 中,避免创建第二套用户、角色、部门和权限系统。 +修改 KRA 既有系统表和系统行为的内容仍放在现有 `internal/biz/system`、`internal/data/system`、`internal/service`、`internal/server` 中,避免创建第二套用户、角色、部门和权限系统。 每个业务模块通过 KRA 现有的 `pkg/module` 机制贡献: diff --git a/internal/app/catalog.go b/internal/app/catalog.go index f2ca3b5..4a77b59 100644 --- a/internal/app/catalog.go +++ b/internal/app/catalog.go @@ -4,8 +4,10 @@ package app import ( + integrationmodule "kra/internal/modules/integration" paymentmodule "kra/internal/modules/payment" systemmodule "kra/internal/modules/system" + taskmodule "kra/internal/modules/task" "kra/pkg/module" platformtask "kra/pkg/task" ) @@ -17,6 +19,8 @@ import ( func Catalog() module.Catalog { return module.Catalog{Definitions: []module.Definition{ systemmodule.Definition(), + integrationmodule.Definition(), + taskmodule.Definition(), paymentmodule.Definition(), }} } diff --git a/internal/app/catalog_test.go b/internal/app/catalog_test.go index 7261e6d..988ed06 100644 --- a/internal/app/catalog_test.go +++ b/internal/app/catalog_test.go @@ -10,16 +10,28 @@ import ( platformtask "kra/pkg/task" ) +func TestCatalogMigrationIDsAreUnique(t *testing.T) { + seen := map[string]string{} + for _, definition := range Catalog().Definitions { + for _, step := range definition.Migrations { + if previous, exists := seen[step.ID]; exists { + t.Fatalf("migration %q is declared by both %s and %s", step.ID, previous, definition.Name) + } + seen[step.ID] = definition.Name + } + } +} + func TestCatalogIncludesSystemDefinition(t *testing.T) { catalog := Catalog() - if len(catalog.Definitions) != 2 { - t.Fatalf("definitions = %d, want 2", len(catalog.Definitions)) + if len(catalog.Definitions) != 4 { + t.Fatalf("definitions = %d, want 4", len(catalog.Definitions)) } - if catalog.Definitions[0].Name != "system" || catalog.Definitions[1].Name != "payment" { - t.Fatalf("definition order = [%q, %q], want [system, payment]", catalog.Definitions[0].Name, catalog.Definitions[1].Name) + if catalog.Definitions[0].Name != "system" || catalog.Definitions[1].Name != "integration" || catalog.Definitions[2].Name != "task" || catalog.Definitions[3].Name != "payment" { + t.Fatalf("definition order = [%q, %q, %q, %q], want [system, integration, task, payment]", catalog.Definitions[0].Name, catalog.Definitions[1].Name, catalog.Definitions[2].Name, catalog.Definitions[3].Name) } - if got := catalog.MigrationSteps(); len(got) != 5 { - t.Fatalf("module migrations = %d, want 5", len(got)) + if got := catalog.MigrationSteps(); len(got) != 8 { + t.Fatalf("module migrations = %d, want 8", len(got)) } if surface := catalog.Surface(); len(surface.Menus) != 3 || len(surface.APIs) != 15 { t.Fatalf("admin surface = %d menus/%d APIs, want 3/15", len(surface.Menus), len(surface.APIs)) diff --git a/internal/biz/system/maintenance.go b/internal/biz/system/maintenance.go new file mode 100644 index 0000000..d0b9301 --- /dev/null +++ b/internal/biz/system/maintenance.go @@ -0,0 +1,19 @@ +package system + +import "context" + +// MaintenanceRepo owns retention work for system-owned tables. Other data +// modules clean their own tables and worker composes the operations. +type MaintenanceRepo interface { + CleanupExpired(context.Context) error +} + +type MaintenanceUsecase struct{ repo MaintenanceRepo } + +func NewMaintenanceUsecase(repo MaintenanceRepo) *MaintenanceUsecase { + return &MaintenanceUsecase{repo: repo} +} + +func (uc *MaintenanceUsecase) CleanupExpired(ctx context.Context) error { + return uc.repo.CleanupExpired(ctx) +} diff --git a/internal/biz/system/provider.go b/internal/biz/system/provider.go index 722d2b6..a4ac350 100644 --- a/internal/biz/system/provider.go +++ b/internal/biz/system/provider.go @@ -27,4 +27,5 @@ var ProviderSet = wire.NewSet( NewMediaUsecase, NewAnnouncementUsecase, NewEmailUsecase, + NewMaintenanceUsecase, ) diff --git a/internal/data/README.md b/internal/data/README.md index 7290cb3..0c4794c 100644 --- a/internal/data/README.md +++ b/internal/data/README.md @@ -3,12 +3,29 @@ `data` owns database clients, persistence models, migrations, configuration watching, and repository implementations. -- `repository/`: system repositories and table persistence -- `payment/`: payment configuration and payment-order persistence +- `system/`: system repositories and system table persistence +- `task/`: timed-task tables and task persistence +- `integration/`: `sys_integration_configs` and integration configuration persistence +- `payment/`: payment-order persistence - each subpackage owns its Wire `ProviderSet`; the root package only binds the - shared `Data` infrastructure and aggregates those sets -- root files: shared database lifecycle, runtime clients, integration-config - storage, data-scope auditing, and migration orchestration + shared `Data` infrastructure and aggregates those sets, mirroring `biz` +- root files: shared database lifecycle, runtime clients, configuration + orchestration, data-scope auditing, and migration orchestration + +新增数据模块时创建 `internal/data/`,提供 `ProviderSet()` 和 +`Migrations()`,再在 `internal/modules//definition.go` 注册迁移与 +管理面,最后在根 `data.ProviderSet` 中注册该模块。根 `data` 不应直接 +拥有业务表 PO,也不应让一个模块引用另一个模块的私有 PO。 Root files intentionally stay in one package because they share `Data` state and reload locks. Do not split them into packages only to reduce file count. + +当前内置数据模块为 `system`、`integration`、`task` 和 `payment`。其中: + +- `system` 只拥有 `sys_*` 系统表、系统仓储、种子和系统维护清理; +- `integration` 唯一拥有 `sys_integration_configs` 表模型、配置仓储和通信默认值; +- `task` 只拥有定时任务与任务日志表; +- `payment` 只拥有 `pay_orders` 等支付持久化,读取集成配置时复用 integration 的表模型。 + +配置文件迁移、数据库切换、Redis/Mongo/对象存储重载仍属于根 data 的生命周期编排, +不等同于某个业务模块的表仓储。 diff --git a/internal/data/data.go b/internal/data/data.go index 9c965dc..00568e2 100644 --- a/internal/data/data.go +++ b/internal/data/data.go @@ -12,8 +12,10 @@ import ( "github.com/redis/go-redis/v9" "gorm.io/gorm" "kra/internal/conf" + dataintegration "kra/internal/data/integration" datapayment "kra/internal/data/payment" - datasystem "kra/internal/data/repository" + datasystem "kra/internal/data/system" + datatask "kra/internal/data/task" "kra/internal/integration/runtimeconfig" "kra/internal/integration/storage" "kra/pkg/module" @@ -24,8 +26,12 @@ var ProviderSet = wire.NewSet( NewIntegrationRuntime, wire.Bind(new(datasystem.Provider), new(*Data)), wire.Bind(new(datasystem.DatabaseProvider), new(*Data)), + wire.Bind(new(dataintegration.Provider), new(*Data)), + wire.Bind(new(datatask.Provider), new(*Data)), wire.Bind(new(datapayment.Provider), new(*Data)), datasystem.ProviderSet, + dataintegration.ProviderSet, + datatask.ProviderSet, datapayment.ProviderSet, ) @@ -232,7 +238,7 @@ func NewData(runtime *conf.Runtime, appLogger *slog.Logger, storageManager *stor if storageManager != nil { storageManager.Replace(activeStorage) } - if db.Migrator().HasTable(&integrationConfigPO{}) { + if db.Migrator().HasTable("sys_integration_configs") { if removeErr := d.removeIntegrationConfigFromFile(); removeErr != nil { appLogger.Warn("remove legacy integration configuration from file", "mod", "integration", "error", removeErr) } diff --git a/internal/data/data_scope_record.go b/internal/data/data_scope_record.go index c46be18..f04e85f 100644 --- a/internal/data/data_scope_record.go +++ b/internal/data/data_scope_record.go @@ -1,22 +1,9 @@ package data import ( - "time" - - "gorm.io/gorm" + datasystem "kra/internal/data/system" ) // dataAccessLogPO is the infrastructure-side write model used by GORM // callbacks. The system module owns the query repository for the same table. -type dataAccessLogPO struct { - ID uint `gorm:"primaryKey"` - CreatedAt time.Time - UpdatedAt time.Time - DeletedAt gorm.DeletedAt `gorm:"index"` - EventType, TargetTable, Operation string - UserID, AuthorityID uint - Scope int - RequestID, Method, Path, Detail string -} - -func (dataAccessLogPO) TableName() string { return "sys_data_access_logs" } +type dataAccessLogPO = datasystem.DataAccessLogPO diff --git a/internal/data/repository/integration_config.go b/internal/data/integration/integration_config.go similarity index 91% rename from internal/data/repository/integration_config.go rename to internal/data/integration/integration_config.go index 510147c..9f0dbc5 100644 --- a/internal/data/repository/integration_config.go +++ b/internal/data/integration/integration_config.go @@ -1,4 +1,4 @@ -package system +package integration import ( "context" @@ -13,7 +13,7 @@ import ( "gorm.io/gorm" ) -type integrationConfigPO struct { +type ConfigPO struct { ID uint `gorm:"primaryKey"` CreatedAt time.Time UpdatedAt time.Time @@ -23,7 +23,7 @@ type integrationConfigPO struct { Config string `gorm:"type:text;not null"` } -func (integrationConfigPO) TableName() string { return "sys_integration_configs" } +func (ConfigPO) TableName() string { return "sys_integration_configs" } type integrationConfigRepo struct{ data Provider } @@ -36,7 +36,7 @@ func NewIntegrationConfigRepo(data Provider) integrationbiz.IntegrationConfigRep } func (r *integrationConfigRepo) ListIntegrationConfigs(ctx context.Context, kind string) ([]*integrationbiz.IntegrationConfig, error) { - var rows []integrationConfigPO + var rows []ConfigPO if err := r.data.DB().WithContext(ctx).Where("kind = ?", kind).Order("provider ASC").Find(&rows).Error; err != nil { return nil, err } @@ -48,7 +48,7 @@ func (r *integrationConfigRepo) ListIntegrationConfigs(ctx context.Context, kind } func (r *integrationConfigRepo) FindIntegrationConfig(ctx context.Context, kind, provider string) (*integrationbiz.IntegrationConfig, error) { - var row integrationConfigPO + var row ConfigPO if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", kind, provider).First(&row).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, errors.New("集成配置不存在") @@ -60,7 +60,7 @@ func (r *integrationConfigRepo) FindIntegrationConfig(ctx context.Context, kind, func (r *integrationConfigRepo) SaveIntegrationConfig(ctx context.Context, config *integrationbiz.IntegrationConfig) error { db := r.data.DB().WithContext(ctx) - var row integrationConfigPO + var row ConfigPO err := db.Where("kind = ? AND provider = ?", config.Kind, config.Provider).First(&row).Error values := integrationObject(config.Values) if errors.Is(err, gorm.ErrRecordNotFound) { @@ -70,7 +70,7 @@ func (r *integrationConfigRepo) SaveIntegrationConfig(ctx context.Context, confi } } encoded, _ := json.Marshal(values) - if err := db.Create(&integrationConfigPO{Kind: config.Kind, Provider: config.Provider, Enabled: config.Enabled, Config: string(encoded)}).Error; err != nil { + if err := db.Create(&ConfigPO{Kind: config.Kind, Provider: config.Provider, Enabled: config.Enabled, Config: string(encoded)}).Error; err != nil { return err } r.publish(config.Kind, config.Provider, config.Enabled, encoded) @@ -94,7 +94,7 @@ func (r *integrationConfigRepo) SaveIntegrationConfig(ctx context.Context, confi } func (r *integrationConfigRepo) DeleteIntegrationConfig(ctx context.Context, kind, provider string) error { - if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", kind, provider).Delete(&integrationConfigPO{}).Error; err != nil { + if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", kind, provider).Delete(&ConfigPO{}).Error; err != nil { return err } if runtime := integrationRuntime(r.data); runtime != nil { @@ -116,7 +116,7 @@ func integrationRuntime(provider Provider) *runtimeconfig.Store { return nil } -func integrationConfigFromPO(row integrationConfigPO) *integrationbiz.IntegrationConfig { +func integrationConfigFromPO(row ConfigPO) *integrationbiz.IntegrationConfig { values := integrationObject(json.RawMessage(row.Config)) maskIntegrationSecrets(row.Kind, row.Provider, values) encoded, _ := json.Marshal(values) diff --git a/internal/data/repository/integration_config_runtime_test.go b/internal/data/integration/integration_config_runtime_test.go similarity index 95% rename from internal/data/repository/integration_config_runtime_test.go rename to internal/data/integration/integration_config_runtime_test.go index 7213911..e45f943 100644 --- a/internal/data/repository/integration_config_runtime_test.go +++ b/internal/data/integration/integration_config_runtime_test.go @@ -1,4 +1,4 @@ -package system +package integration import ( "context" @@ -21,7 +21,7 @@ func TestIntegrationConfigSavePublishesUnmaskedRuntimeValues(t *testing.T) { if err != nil { t.Fatal(err) } - if err = db.AutoMigrate(&integrationConfigPO{}); err != nil { + if err = db.AutoMigrate(&ConfigPO{}); err != nil { t.Fatal(err) } provider := &integrationRuntimeTestProvider{Data: &Data{gormDB: newReloadableDB(db, nil)}, store: runtimeconfig.NewStore()} diff --git a/internal/data/integration/migrations.go b/internal/data/integration/migrations.go new file mode 100644 index 0000000..61d81cd --- /dev/null +++ b/internal/data/integration/migrations.go @@ -0,0 +1,45 @@ +package integration + +import ( + "encoding/json" + + integrationbiz "kra/internal/biz/integration" + "kra/pkg/database/migration" + + "gorm.io/gorm" +) + +func Migrations() []migration.Step { + return []migration.Step{ + {ID: "202608200001_data_infrastructure", Migrate: func(db *gorm.DB) error { + return migration.CreateMissingTables(db, &ConfigPO{}) + }}, + {ID: "202608210001_communication_integration_defaults", Migrate: ensureCommunicationIntegrationConfigs}, + } +} + +func ensureCommunicationIntegrationConfigs(db *gorm.DB) error { + defaults := []struct{ kind, provider string }{ + {integrationbiz.IntegrationKindMQ, "emqx"}, + {integrationbiz.IntegrationKindMQ, "rabbitmq"}, + {integrationbiz.IntegrationKindWebSocket, "melody"}, + } + for _, item := range defaults { + var row ConfigPO + err := db.Where("kind = ? AND provider = ?", item.kind, item.provider).First(&row).Error + if gorm.ErrRecordNotFound == err { + values, marshalErr := json.Marshal(integrationbiz.DefaultIntegrationConfig(item.kind, item.provider)) + if marshalErr != nil { + return marshalErr + } + if err = db.Create(&ConfigPO{Kind: item.kind, Provider: item.provider, Enabled: false, Config: string(values)}).Error; err != nil { + return err + } + continue + } + if err != nil { + return err + } + } + return nil +} diff --git a/internal/data/integration/provider.go b/internal/data/integration/provider.go new file mode 100644 index 0000000..c80c04c --- /dev/null +++ b/internal/data/integration/provider.go @@ -0,0 +1,12 @@ +package integration + +import ( + "gorm.io/gorm" + "kra/internal/integration/runtimeconfig" +) + +// Provider is the minimal database/runtime seam for integration configuration. +type Provider interface { + DB() *gorm.DB + IntegrationRuntime() *runtimeconfig.Store +} diff --git a/internal/data/integration/providers.go b/internal/data/integration/providers.go new file mode 100644 index 0000000..a862518 --- /dev/null +++ b/internal/data/integration/providers.go @@ -0,0 +1,5 @@ +package integration + +import "github.com/google/wire" + +var ProviderSet = wire.NewSet(NewIntegrationConfigRepo) diff --git a/internal/data/integration/runtime.go b/internal/data/integration/runtime.go new file mode 100644 index 0000000..2799185 --- /dev/null +++ b/internal/data/integration/runtime.go @@ -0,0 +1,29 @@ +package integration + +import ( + "errors" + + "kra/internal/integration/runtimeconfig" + + "gorm.io/gorm" +) + +// ReadRuntime returns the communication integration snapshot used by the +// long-lived MQ/WebSocket adapters. +func ReadRuntime(db *gorm.DB) ([]runtimeconfig.Config, error) { + if db == nil || !db.Migrator().HasTable(&ConfigPO{}) { + return nil, nil + } + var rows []ConfigPO + if err := db.Session(&gorm.Session{NewDB: true}). + Where("kind IN ?", []string{"mq", "websocket"}). + Order("kind ASC, provider ASC"). + Find(&rows).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, err + } + configs := make([]runtimeconfig.Config, 0, len(rows)) + for _, row := range rows { + configs = append(configs, runtimeconfig.Config{Kind: row.Kind, Provider: row.Provider, Enabled: row.Enabled, Values: []byte(row.Config)}) + } + return configs, nil +} diff --git a/internal/data/integration/testing_support_test.go b/internal/data/integration/testing_support_test.go new file mode 100644 index 0000000..32c7897 --- /dev/null +++ b/internal/data/integration/testing_support_test.go @@ -0,0 +1,42 @@ +package integration + +import ( + "context" + "fmt" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + "kra/internal/integration/runtimeconfig" +) + +type Data struct { + gormDB *reloadableDB + store *runtimeconfig.Store +} + +func (d *Data) DB() *gorm.DB { + if d == nil || d.gormDB == nil { + return nil + } + return d.gormDB.DB() +} + +func (d *Data) IntegrationRuntime() *runtimeconfig.Store { return d.store } + +type reloadableDB struct{ db *gorm.DB } + +func newReloadableDB(db *gorm.DB, _ ...any) *reloadableDB { return &reloadableDB{db: db} } +func (r *reloadableDB) DB() *gorm.DB { + if r == nil { + return nil + } + return r.db +} +func (r *reloadableDB) WithContext(ctx context.Context) *gorm.DB { return r.db.WithContext(ctx) } + +func openWithDriver(driver, dsn string) (*gorm.DB, error) { + if driver != "sqlite" { + return nil, fmt.Errorf("unsupported test database driver %q", driver) + } + return gorm.Open(sqlite.Open(dsn), &gorm.Config{}) +} diff --git a/internal/data/integration_config.go b/internal/data/integration_config.go index e5e59c7..980ad8d 100644 --- a/internal/data/integration_config.go +++ b/internal/data/integration_config.go @@ -6,9 +6,9 @@ import ( "errors" "fmt" "strings" - "time" "kra/internal/conf" + dataintegration "kra/internal/data/integration" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" @@ -21,21 +21,6 @@ const ( integrationKindPayment = "payment" ) -// integrationConfigPO stores credentials and provider-specific options for -// external services. Payment integrations use the same table with kind -// "payment", keeping secrets out of the bootstrap configuration file. -type integrationConfigPO struct { - ID uint `gorm:"primaryKey"` - CreatedAt time.Time - UpdatedAt time.Time - Kind string `gorm:"size:32;not null;uniqueIndex:idx_integration_kind_provider"` - Provider string `gorm:"size:64;not null;uniqueIndex:idx_integration_kind_provider"` - Enabled bool `gorm:"not null;default:false;index"` - Config string `gorm:"type:text;not null"` -} - -func (integrationConfigPO) TableName() string { return "sys_integration_configs" } - var storageProviderNames = []string{ "local", "qiniu", @@ -140,11 +125,11 @@ func saveStorageIntegrationConfig(db *gorm.DB, storage *conf.AdminBackend_Storag if err != nil { return fmt.Errorf("encode %s integration configuration: %w", provider, err) } - var current integrationConfigPO + var current dataintegration.ConfigPO err = tx.Where("kind = ? AND provider = ?", integrationKindStorage, provider).First(¤t).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): - current = integrationConfigPO{Kind: integrationKindStorage, Provider: provider} + current = dataintegration.ConfigPO{Kind: integrationKindStorage, Provider: provider} current.Enabled, current.Config = provider == active, value if err = tx.Create(¤t).Error; err != nil { return err @@ -162,7 +147,7 @@ func saveStorageIntegrationConfig(db *gorm.DB, storage *conf.AdminBackend_Storag } func loadStorageIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Storage, bool, error) { - var rows []integrationConfigPO + var rows []dataintegration.ConfigPO err := db.Session(&gorm.Session{NewDB: true}). Where("kind = ?", integrationKindStorage). Order("id ASC"). @@ -191,7 +176,7 @@ func loadStorageIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Storage, bool // sole source of truth. func resolveStorageIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_Storage) (*conf.AdminBackend_Storage, error) { clean := db.Session(&gorm.Session{NewDB: true}) - if !clean.Migrator().HasTable(&integrationConfigPO{}) { + if !clean.Migrator().HasTable(&dataintegration.ConfigPO{}) { if legacy == nil { return &conf.AdminBackend_Storage{Type: "local"}, nil } @@ -219,7 +204,7 @@ func (d *Data) persistStorageIntegrationConfig(ctx context.Context, storage *con return errors.New("database is not initialized") } db := d.gormDB.WithContext(ctx) - if !db.Migrator().HasTable(&integrationConfigPO{}) { + if !db.Migrator().HasTable(&dataintegration.ConfigPO{}) { return errors.New("integration configuration table does not exist") } return saveStorageIntegrationConfig(db, storage) @@ -239,11 +224,11 @@ func saveEmailIntegrationConfig(db *gorm.DB, email *conf.AdminBackend_Email) err } enabled := email.Host != "" && email.From != "" && email.Secret != "" && email.Port > 0 clean := db.Session(&gorm.Session{NewDB: true}) - var current integrationConfigPO + var current dataintegration.ConfigPO err = clean.Where("kind = ? AND provider = ?", integrationKindEmail, "smtp").First(¤t).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): - return clean.Create(&integrationConfigPO{ + return clean.Create(&dataintegration.ConfigPO{ Kind: integrationKindEmail, Provider: "smtp", Enabled: enabled, Config: string(raw), }).Error case err != nil: @@ -254,7 +239,7 @@ func saveEmailIntegrationConfig(db *gorm.DB, email *conf.AdminBackend_Email) err } func loadEmailIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Email, bool, error) { - var row integrationConfigPO + var row dataintegration.ConfigPO err := db.Session(&gorm.Session{NewDB: true}). Where("kind = ? AND provider = ?", integrationKindEmail, "smtp"). First(&row).Error @@ -276,7 +261,7 @@ func loadEmailIntegrationConfig(db *gorm.DB) (*conf.AdminBackend_Email, bool, er func resolveEmailIntegrationConfig(db *gorm.DB, legacy *conf.AdminBackend_Email) (*conf.AdminBackend_Email, error) { clean := db.Session(&gorm.Session{NewDB: true}) - if !clean.Migrator().HasTable(&integrationConfigPO{}) { + if !clean.Migrator().HasTable(&dataintegration.ConfigPO{}) { if legacy == nil { return defaultEmailIntegrationConfig(), nil } @@ -304,7 +289,7 @@ func (d *Data) persistEmailIntegrationConfig(ctx context.Context, email *conf.Ad return errors.New("database is not initialized") } db := d.gormDB.WithContext(ctx) - if !db.Migrator().HasTable(&integrationConfigPO{}) { + if !db.Migrator().HasTable(&dataintegration.ConfigPO{}) { return errors.New("integration configuration table does not exist") } return saveEmailIntegrationConfig(db, email) diff --git a/internal/data/integration_config_test.go b/internal/data/integration_config_test.go index 8a8634d..8f974b1 100644 --- a/internal/data/integration_config_test.go +++ b/internal/data/integration_config_test.go @@ -26,7 +26,7 @@ func openIntegrationConfigTestDB(t *testing.T) *gorm.DB { t.Fatal(err) } t.Cleanup(func() { _ = sqlDB.Close() }) - if err = db.AutoMigrate(&integrationConfigPO{}); err != nil { + if err = db.AutoMigrate(&dataintegration.ConfigPO{}); err != nil { t.Fatal(err) } return db @@ -42,10 +42,10 @@ func TestMigrateAllCreatesIntegrationConfigTable(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = sqlDB.Close() }) - if err = migrateAll(db); err != nil { + if err = migrateAll(db, testCatalog()); err != nil { t.Fatal(err) } - if !db.Migrator().HasTable(&integrationConfigPO{}) { + if !db.Migrator().HasTable(&dataintegration.ConfigPO{}) { t.Fatal("migrateAll did not create sys_integration_configs") } } @@ -73,7 +73,7 @@ func TestStorageIntegrationConfigRoundTrip(t *testing.T) { if err := saveEmailIntegrationConfig(db, email); err != nil { t.Fatal(err) } - if err := db.Create(&integrationConfigPO{Kind: integrationKindPayment, Provider: "wechat-pay", Config: `{"merchant_id":"123"}`}).Error; err != nil { + if err := db.Create(&dataintegration.ConfigPO{Kind: integrationKindPayment, Provider: "wechat-pay", Config: `{"merchant_id":"123"}`}).Error; err != nil { t.Fatal(err) } @@ -103,13 +103,13 @@ func TestStorageIntegrationConfigRoundTrip(t *testing.T) { } var storageCount, emailCount, paymentCount int64 - if err = db.Model(&integrationConfigPO{}).Where("kind = ?", integrationKindStorage).Count(&storageCount).Error; err != nil { + if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindStorage).Count(&storageCount).Error; err != nil { t.Fatal(err) } - if err = db.Model(&integrationConfigPO{}).Where("kind = ?", integrationKindPayment).Count(&paymentCount).Error; err != nil { + if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindPayment).Count(&paymentCount).Error; err != nil { t.Fatal(err) } - if err = db.Model(&integrationConfigPO{}).Where("kind = ?", integrationKindEmail).Count(&emailCount).Error; err != nil { + if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindEmail).Count(&emailCount).Error; err != nil { t.Fatal(err) } if storageCount != int64(len(storageProviderNames)) { diff --git a/internal/data/integration_defaults.go b/internal/data/integration_defaults.go deleted file mode 100644 index 0baca30..0000000 --- a/internal/data/integration_defaults.go +++ /dev/null @@ -1,38 +0,0 @@ -package data - -import ( - "encoding/json" - "errors" - integrationbiz "kra/internal/biz/integration" - - "gorm.io/gorm" -) - -func ensureCommunicationIntegrationConfigs(db *gorm.DB) error { - defaults := []struct { - kind string - provider string - }{ - {kind: integrationbiz.IntegrationKindMQ, provider: "emqx"}, - {kind: integrationbiz.IntegrationKindMQ, provider: "rabbitmq"}, - {kind: integrationbiz.IntegrationKindWebSocket, provider: "melody"}, - } - for _, item := range defaults { - var row integrationConfigPO - err := db.Where("kind = ? AND provider = ?", item.kind, item.provider).First(&row).Error - if errors.Is(err, gorm.ErrRecordNotFound) { - values, marshalErr := json.Marshal(integrationbiz.DefaultIntegrationConfig(item.kind, item.provider)) - if marshalErr != nil { - return marshalErr - } - if err = db.Create(&integrationConfigPO{Kind: item.kind, Provider: item.provider, Enabled: false, Config: string(values)}).Error; err != nil { - return err - } - continue - } - if err != nil { - return err - } - } - return nil -} diff --git a/internal/data/integration_runtime.go b/internal/data/integration_runtime.go index 0646d4b..eec7ab5 100644 --- a/internal/data/integration_runtime.go +++ b/internal/data/integration_runtime.go @@ -1,29 +1,13 @@ package data import ( - "errors" - - "kra/internal/integration/runtimeconfig" - "gorm.io/gorm" + dataintegration "kra/internal/data/integration" + "kra/internal/integration/runtimeconfig" ) func readIntegrationRuntime(db *gorm.DB) ([]runtimeconfig.Config, error) { - if db == nil || !db.Migrator().HasTable(&integrationConfigPO{}) { - return nil, nil - } - var rows []integrationConfigPO - if err := db.Session(&gorm.Session{NewDB: true}). - Where("kind IN ?", []string{"mq", "websocket"}). - Order("kind ASC, provider ASC"). - Find(&rows).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { - return nil, err - } - configs := make([]runtimeconfig.Config, 0, len(rows)) - for _, row := range rows { - configs = append(configs, runtimeconfig.Config{Kind: row.Kind, Provider: row.Provider, Enabled: row.Enabled, Values: []byte(row.Config)}) - } - return configs, nil + return dataintegration.ReadRuntime(db) } func (d *Data) loadIntegrationRuntime(db *gorm.DB) error { diff --git a/internal/data/migrations.go b/internal/data/migrations.go index 42b16ed..4140eee 100644 --- a/internal/data/migrations.go +++ b/internal/data/migrations.go @@ -1,8 +1,6 @@ package data import ( - datapayment "kra/internal/data/payment" - datasystem "kra/internal/data/repository" "kra/pkg/database/migration" "kra/pkg/module" @@ -10,26 +8,14 @@ import ( ) func InfrastructureMigrations() []migration.Step { - return []migration.Step{ - { - ID: "202608200001_data_infrastructure", - Migrate: func(db *gorm.DB) error { - return migration.CreateMissingTables(db, &integrationConfigPO{}) - }, - }, - {ID: "202608210001_communication_integration_defaults", Migrate: ensureCommunicationIntegrationConfigs}, - } + return nil } -// migrateAll is the single data-layer migration entry point. Module-specific -// schema work stays with the module that owns its persistent objects. -func migrateAll(db *gorm.DB, catalogs ...module.Catalog) error { - steps := InfrastructureMigrations() - if len(catalogs) > 0 && len(catalogs[0].MigrationSteps()) > 0 { - steps = append(steps, catalogs[0].MigrationSteps()...) - } else { - steps = append(steps, datasystem.Migrations()...) - steps = append(steps, datapayment.Migrations()...) - } +// migrateAll is the single data-layer migration entry point. Every module +// must register its migrations through the application catalog; there is no +// hidden system/payment fallback that could silently omit a new module. +func migrateAll(db *gorm.DB, catalog module.Catalog) error { + steps := append([]migration.Step{}, InfrastructureMigrations()...) + steps = append(steps, catalog.MigrationSteps()...) return migration.Run(db, steps) } diff --git a/internal/data/migrations_test.go b/internal/data/migrations_test.go index 34a89d9..f219ee4 100644 --- a/internal/data/migrations_test.go +++ b/internal/data/migrations_test.go @@ -3,18 +3,29 @@ package data import ( "testing" + integrationmodule "kra/internal/modules/integration" + paymentmodule "kra/internal/modules/payment" + systemmodule "kra/internal/modules/system" + taskmodule "kra/internal/modules/task" "kra/pkg/database/migration" + platformmodule "kra/pkg/module" ) +func testCatalog() platformmodule.Catalog { + return platformmodule.Catalog{Definitions: []platformmodule.Definition{ + systemmodule.Definition(), integrationmodule.Definition(), taskmodule.Definition(), paymentmodule.Definition(), + }} +} + func TestMigrateAllRunsModuleSchemasWithoutBootstrapSeed(t *testing.T) { db, err := openWithDriver("sqlite", "file:"+t.Name()+"?mode=memory&cache=shared") if err != nil { t.Fatal(err) } - if err = migrateAll(db); err != nil { + if err = migrateAll(db, testCatalog()); err != nil { t.Fatal(err) } - if err = migrateAll(db); err != nil { + if err = migrateAll(db, testCatalog()); err != nil { t.Fatalf("second migration run: %v", err) } for _, table := range []string{"sys_integration_configs", "sys_users", "sys_base_menus", "pay_orders"} { @@ -26,10 +37,10 @@ func TestMigrateAllRunsModuleSchemasWithoutBootstrapSeed(t *testing.T) { if err = db.Table(migration.TableName).Count(&versions).Error; err != nil { t.Fatal(err) } - if versions != 7 { - t.Fatalf("migration versions = %d, want 7", versions) + if versions != 8 { + t.Fatalf("migration versions = %d, want 8", versions) } - var communicationRows []integrationConfigPO + var communicationRows []dataintegration.ConfigPO if err = db.Where("kind IN ?", []string{"mq", "websocket"}).Order("kind, provider").Find(&communicationRows).Error; err != nil { t.Fatal(err) } diff --git a/internal/data/payment/models.go b/internal/data/payment/models.go index af0bc97..fa9c1b5 100644 --- a/internal/data/payment/models.go +++ b/internal/data/payment/models.go @@ -1,17 +1,5 @@ package payment -import "time" +import dataintegration "kra/internal/data/integration" const integrationKindPayment = "payment" - -type integrationConfigPO struct { - ID uint `gorm:"primaryKey"` - CreatedAt time.Time - UpdatedAt time.Time - Kind string `gorm:"size:32;not null;uniqueIndex:idx_integration_kind_provider"` - Provider string `gorm:"size:64;not null;uniqueIndex:idx_integration_kind_provider"` - Enabled bool `gorm:"not null;default:false;index"` - Config string `gorm:"type:text;not null"` -} - -func (integrationConfigPO) TableName() string { return "sys_integration_configs" } diff --git a/internal/data/payment/payment.go b/internal/data/payment/payment.go index 0bdf17d..8572ded 100644 --- a/internal/data/payment/payment.go +++ b/internal/data/payment/payment.go @@ -26,12 +26,12 @@ func NewPaymentRepo(data Provider) bizpayment.PaymentRepo { return &paymentRepo{ func ensurePaymentIntegrationConfigs(db *gorm.DB) error { for _, provider := range bizpayment.SupportedPaymentProviders { - var row integrationConfigPO + var row dataintegration.ConfigPO err := db.Where("kind = ? AND provider = ?", integrationKindPayment, provider).First(&row).Error defaults := integrationbiz.DefaultIntegrationConfig(integrationKindPayment, provider) if errors.Is(err, gorm.ErrRecordNotFound) { encoded, _ := json.Marshal(defaults) - if err := db.Create(&integrationConfigPO{Kind: integrationKindPayment, Provider: provider, Enabled: false, Config: string(encoded)}).Error; err != nil { + if err := db.Create(&dataintegration.ConfigPO{Kind: integrationKindPayment, Provider: provider, Enabled: false, Config: string(encoded)}).Error; err != nil { return err } continue @@ -58,8 +58,8 @@ func ensurePaymentIntegrationConfigs(db *gorm.DB) error { return nil } -func (r *paymentRepo) row(ctx context.Context, provider string) (*integrationConfigPO, map[string]any, error) { - var row integrationConfigPO +func (r *paymentRepo) row(ctx context.Context, provider string) (*dataintegration.ConfigPO, map[string]any, error) { + var row dataintegration.ConfigPO if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationKindPayment, provider).First(&row).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil, bizpayment.ErrPaymentProviderNotFound @@ -298,7 +298,7 @@ func recordPaymentTestError(ctx context.Context, data Provider, provider, tradeN } func (r *paymentRepo) testRow(ctx context.Context, provider string) (map[string]any, error) { - var row integrationConfigPO + var row dataintegration.ConfigPO if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationKindPayment, provider).First(&row).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, bizpayment.ErrPaymentProviderNotFound diff --git a/internal/data/payment/testing_support_test.go b/internal/data/payment/testing_support_test.go index 245efa6..f56b6e8 100644 --- a/internal/data/payment/testing_support_test.go +++ b/internal/data/payment/testing_support_test.go @@ -36,7 +36,7 @@ func openIntegrationConfigTestDB(t *testing.T) *gorm.DB { if err != nil { t.Fatal(err) } - if err := db.AutoMigrate(&integrationConfigPO{}); err != nil { + if err := db.AutoMigrate(&dataintegration.ConfigPO{}); err != nil { t.Fatal(err) } t.Cleanup(func() { @@ -48,7 +48,7 @@ func openIntegrationConfigTestDB(t *testing.T) *gorm.DB { } func migrateAll(db *gorm.DB) error { - if err := db.AutoMigrate(&integrationConfigPO{}); err != nil { + if err := db.AutoMigrate(&dataintegration.ConfigPO{}); err != nil { return err } for _, step := range Migrations() { diff --git a/internal/data/repository/announcement.go b/internal/data/system/announcement.go similarity index 100% rename from internal/data/repository/announcement.go rename to internal/data/system/announcement.go diff --git a/internal/data/repository/announcement_test.go b/internal/data/system/announcement_test.go similarity index 100% rename from internal/data/repository/announcement_test.go rename to internal/data/system/announcement_test.go diff --git a/internal/data/repository/api.go b/internal/data/system/api.go similarity index 100% rename from internal/data/repository/api.go rename to internal/data/system/api.go diff --git a/internal/data/repository/api_policy.go b/internal/data/system/api_policy.go similarity index 100% rename from internal/data/repository/api_policy.go rename to internal/data/system/api_policy.go diff --git a/internal/data/repository/api_policy_test.go b/internal/data/system/api_policy_test.go similarity index 100% rename from internal/data/repository/api_policy_test.go rename to internal/data/system/api_policy_test.go diff --git a/internal/data/repository/api_sync.go b/internal/data/system/api_sync.go similarity index 100% rename from internal/data/repository/api_sync.go rename to internal/data/system/api_sync.go diff --git a/internal/data/repository/api_token.go b/internal/data/system/api_token.go similarity index 100% rename from internal/data/repository/api_token.go rename to internal/data/system/api_token.go diff --git a/internal/data/repository/audit.go b/internal/data/system/audit.go similarity index 100% rename from internal/data/repository/audit.go rename to internal/data/system/audit.go diff --git a/internal/data/repository/audit_delete_test.go b/internal/data/system/audit_delete_test.go similarity index 100% rename from internal/data/repository/audit_delete_test.go rename to internal/data/system/audit_delete_test.go diff --git a/internal/data/repository/authority.go b/internal/data/system/authority.go similarity index 100% rename from internal/data/repository/authority.go rename to internal/data/system/authority.go diff --git a/internal/data/repository/authority_test.go b/internal/data/system/authority_test.go similarity index 100% rename from internal/data/repository/authority_test.go rename to internal/data/system/authority_test.go diff --git a/internal/data/repository/bootstrap.go b/internal/data/system/bootstrap.go similarity index 100% rename from internal/data/repository/bootstrap.go rename to internal/data/system/bootstrap.go diff --git a/internal/data/repository/casbin.go b/internal/data/system/casbin.go similarity index 100% rename from internal/data/repository/casbin.go rename to internal/data/system/casbin.go diff --git a/internal/data/repository/data_access_log.go b/internal/data/system/data_access_log.go similarity index 83% rename from internal/data/repository/data_access_log.go rename to internal/data/system/data_access_log.go index 1e6da45..994db67 100644 --- a/internal/data/repository/data_access_log.go +++ b/internal/data/system/data_access_log.go @@ -10,7 +10,7 @@ import ( "gorm.io/gorm" ) -type dataAccessLogPO struct { +type DataAccessLogPO struct { ID uint `gorm:"primaryKey"` CreatedAt time.Time UpdatedAt time.Time @@ -21,16 +21,16 @@ type dataAccessLogPO struct { RequestID, Method, Path, Detail string } -func (dataAccessLogPO) TableName() string { return "sys_data_access_logs" } +func (DataAccessLogPO) TableName() string { return "sys_data_access_logs" } func (r *auditRecorderRepo) RecordDataAccess(ctx context.Context, v *system.DataAccessLog) error { - return r.data.DB().WithContext(ctx).Create(&dataAccessLogPO{EventType: v.EventType, TargetTable: v.TargetTable, Operation: v.Operation, UserID: v.UserID, AuthorityID: v.AuthorityID, Scope: v.Scope, RequestID: v.RequestID, Method: v.Method, Path: v.Path, Detail: v.Detail}).Error + return r.data.DB().WithContext(ctx).Create(&DataAccessLogPO{EventType: v.EventType, TargetTable: v.TargetTable, Operation: v.Operation, UserID: v.UserID, AuthorityID: v.AuthorityID, Scope: v.Scope, RequestID: v.RequestID, Method: v.Method, Path: v.Path, Detail: v.Detail}).Error } -func dataAccessFromPO(v dataAccessLogPO) *system.DataAccessLog { +func dataAccessFromPO(v DataAccessLogPO) *system.DataAccessLog { return &system.DataAccessLog{ID: v.ID, CreatedAt: v.CreatedAt, UpdatedAt: v.UpdatedAt, EventType: v.EventType, TargetTable: v.TargetTable, Operation: v.Operation, UserID: v.UserID, AuthorityID: v.AuthorityID, Scope: v.Scope, RequestID: v.RequestID, Method: v.Method, Path: v.Path, Detail: v.Detail} } func (r *auditQueryRepo) ListDataAccess(ctx context.Context, page, size int, q *system.DataAccessLog) ([]*system.DataAccessLog, int64, error) { - db := r.data.DB().WithContext(ctx).Model(&dataAccessLogPO{}) + db := r.data.DB().WithContext(ctx).Model(&DataAccessLogPO{}) if q != nil { if q.EventType != "" { db = db.Where("event_type = ?", q.EventType) @@ -43,7 +43,7 @@ func (r *auditQueryRepo) ListDataAccess(ctx context.Context, page, size int, q * if err := db.Count(&total).Error; err != nil { return nil, 0, err } - var pos []dataAccessLogPO + var pos []DataAccessLogPO if err := pagination.ApplyRequired(db.Order("id desc"), page, size, 100).Find(&pos).Error; err != nil { return nil, 0, err } @@ -54,5 +54,5 @@ func (r *auditQueryRepo) ListDataAccess(ctx context.Context, page, size int, q * return out, total, nil } func (r *auditQueryRepo) DeleteDataAccess(ctx context.Context, ids []uint) error { - return r.data.DB().WithContext(ctx).Delete(&dataAccessLogPO{}, "id IN ?", ids).Error + return r.data.DB().WithContext(ctx).Delete(&DataAccessLogPO{}, "id IN ?", ids).Error } diff --git a/internal/data/repository/department.go b/internal/data/system/department.go similarity index 100% rename from internal/data/repository/department.go rename to internal/data/system/department.go diff --git a/internal/data/repository/dictionary.go b/internal/data/system/dictionary.go similarity index 100% rename from internal/data/repository/dictionary.go rename to internal/data/system/dictionary.go diff --git a/internal/data/repository/dictionary_department_parity_test.go b/internal/data/system/dictionary_department_parity_test.go similarity index 100% rename from internal/data/repository/dictionary_department_parity_test.go rename to internal/data/system/dictionary_department_parity_test.go diff --git a/internal/data/repository/error_record.go b/internal/data/system/error_record.go similarity index 100% rename from internal/data/repository/error_record.go rename to internal/data/system/error_record.go diff --git a/internal/data/repository/error_record_test.go b/internal/data/system/error_record_test.go similarity index 100% rename from internal/data/repository/error_record_test.go rename to internal/data/system/error_record_test.go diff --git a/internal/data/repository/export.go b/internal/data/system/export.go similarity index 100% rename from internal/data/repository/export.go rename to internal/data/system/export.go diff --git a/internal/data/repository/log_file.go b/internal/data/system/log_file.go similarity index 100% rename from internal/data/repository/log_file.go rename to internal/data/system/log_file.go diff --git a/internal/data/repository/log_file_test.go b/internal/data/system/log_file_test.go similarity index 100% rename from internal/data/repository/log_file_test.go rename to internal/data/system/log_file_test.go diff --git a/internal/data/repository/login_log.go b/internal/data/system/login_log.go similarity index 100% rename from internal/data/repository/login_log.go rename to internal/data/system/login_log.go diff --git a/internal/data/system/maintenance.go b/internal/data/system/maintenance.go new file mode 100644 index 0000000..2b8255e --- /dev/null +++ b/internal/data/system/maintenance.go @@ -0,0 +1,26 @@ +package system + +import ( + "context" + "time" + + bizsystem "kra/internal/biz/system" + + "gorm.io/gorm" +) + +type maintenanceRepo struct{ data Provider } + +func NewMaintenanceRepo(data Provider) bizsystem.MaintenanceRepo { + return &maintenanceRepo{data: data} +} + +func (r *maintenanceRepo) CleanupExpired(ctx context.Context) error { + now := time.Now() + return r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Unscoped().Where("created_at < ?", now.Add(-2160*time.Hour)).Delete(&operationPO{}).Error; err != nil { + return err + } + return tx.Unscoped().Where("created_at < ?", now.Add(-168*time.Hour)).Delete(&jwtBlacklistPO{}).Error + }) +} diff --git a/internal/data/repository/media.go b/internal/data/system/media.go similarity index 100% rename from internal/data/repository/media.go rename to internal/data/system/media.go diff --git a/internal/data/repository/media_test.go b/internal/data/system/media_test.go similarity index 100% rename from internal/data/repository/media_test.go rename to internal/data/system/media_test.go diff --git a/internal/data/repository/media_upload.go b/internal/data/system/media_upload.go similarity index 100% rename from internal/data/repository/media_upload.go rename to internal/data/system/media_upload.go diff --git a/internal/data/repository/menu.go b/internal/data/system/menu.go similarity index 100% rename from internal/data/repository/menu.go rename to internal/data/system/menu.go diff --git a/internal/data/repository/menu_test.go b/internal/data/system/menu_test.go similarity index 100% rename from internal/data/repository/menu_test.go rename to internal/data/system/menu_test.go diff --git a/internal/data/repository/migrations.go b/internal/data/system/migrations.go similarity index 97% rename from internal/data/repository/migrations.go rename to internal/data/system/migrations.go index 34986bc..c5619a2 100644 --- a/internal/data/repository/migrations.go +++ b/internal/data/system/migrations.go @@ -22,8 +22,8 @@ func Migrations() []migration.Step { &departmentPO{}, &positionPO{}, &userDepartmentPO{}, &userPositionPO{}, &authorityDepartmentPO{}, &dictionaryPO{}, &dictionaryDetailPO{}, ¶meterPO{}, &apiTokenPO{}, &jwtBlacklistPO{}, &SecurityConfigPO{}, &versionPO{}, &exportTemplatePO{}, &exportConditionPO{}, &exportJoinPO{}, - &operationPO{}, &loginLogPO{}, &dataAccessLogPO{}, &errorRecordPO{}, - &taskPO{}, &taskLogPO{}, &mediaPO{}, &categoryPO{}, &uploadSessionPO{}, &uploadChunkPO{}, + &operationPO{}, &loginLogPO{}, &DataAccessLogPO{}, &errorRecordPO{}, + &mediaPO{}, &categoryPO{}, &uploadSessionPO{}, &uploadChunkPO{}, &announcementPO{}, ) }, diff --git a/internal/data/repository/migrations_test.go b/internal/data/system/migrations_test.go similarity index 100% rename from internal/data/repository/migrations_test.go rename to internal/data/system/migrations_test.go diff --git a/internal/data/repository/models.go b/internal/data/system/models.go similarity index 100% rename from internal/data/repository/models.go rename to internal/data/system/models.go diff --git a/internal/data/repository/operation_log.go b/internal/data/system/operation_log.go similarity index 100% rename from internal/data/repository/operation_log.go rename to internal/data/system/operation_log.go diff --git a/internal/data/repository/organization_test.go b/internal/data/system/organization_test.go similarity index 100% rename from internal/data/repository/organization_test.go rename to internal/data/system/organization_test.go diff --git a/internal/data/repository/parameter.go b/internal/data/system/parameter.go similarity index 100% rename from internal/data/repository/parameter.go rename to internal/data/system/parameter.go diff --git a/internal/data/repository/parameter_test.go b/internal/data/system/parameter_test.go similarity index 100% rename from internal/data/repository/parameter_test.go rename to internal/data/system/parameter_test.go diff --git a/internal/data/repository/permission.go b/internal/data/system/permission.go similarity index 100% rename from internal/data/repository/permission.go rename to internal/data/system/permission.go diff --git a/internal/data/repository/permission_test.go b/internal/data/system/permission_test.go similarity index 100% rename from internal/data/repository/permission_test.go rename to internal/data/system/permission_test.go diff --git a/internal/data/repository/position.go b/internal/data/system/position.go similarity index 100% rename from internal/data/repository/position.go rename to internal/data/system/position.go diff --git a/internal/data/repository/provider.go b/internal/data/system/provider.go similarity index 100% rename from internal/data/repository/provider.go rename to internal/data/system/provider.go diff --git a/internal/data/repository/providers.go b/internal/data/system/providers.go similarity index 92% rename from internal/data/repository/providers.go rename to internal/data/system/providers.go index 4bd14c6..a5cc858 100644 --- a/internal/data/repository/providers.go +++ b/internal/data/system/providers.go @@ -22,8 +22,7 @@ var ProviderSet = wire.NewSet( NewAuditRepo, NewAuditRecorderRepo, NewLogFileRepo, - NewTaskRepo, NewMediaRepo, NewAnnouncementRepo, - NewIntegrationConfigRepo, + NewMaintenanceRepo, ) diff --git a/internal/data/repository/runtime.go b/internal/data/system/runtime.go similarity index 100% rename from internal/data/repository/runtime.go rename to internal/data/system/runtime.go diff --git a/internal/data/repository/security.go b/internal/data/system/security.go similarity index 100% rename from internal/data/repository/security.go rename to internal/data/system/security.go diff --git a/internal/data/repository/seed.go b/internal/data/system/seed.go similarity index 89% rename from internal/data/repository/seed.go rename to internal/data/system/seed.go index bdef008..45db9f2 100644 --- a/internal/data/repository/seed.go +++ b/internal/data/system/seed.go @@ -14,21 +14,20 @@ import ( ) func SeedSystem(ctx context.Context, db *gorm.DB, input *system.DatabaseConfig, surfaces ...platformmodule.Surface) error { - return seedSystem(ctx, db, input, nil, surfaces...) + return seedSystem(ctx, db, input, surfaces...) } -// SeedSystemWithCatalog applies module-contributed administration surfaces and -// default timed tasks in one transaction. The system module remains the owner -// of the system tables, while other modules contribute through the catalog. +// SeedSystemWithCatalog applies module-contributed administration surfaces in +// one transaction. Each data module seeds its own persistent tables. func SeedSystemWithCatalog(ctx context.Context, db *gorm.DB, input *system.DatabaseConfig, catalog platformmodule.Catalog) error { surfaces := make([]platformmodule.Surface, 0, len(catalog.Definitions)) for _, definition := range catalog.Definitions { surfaces = append(surfaces, definition.Surface) } - return seedSystem(ctx, db, input, catalog.DefaultTimedTasks(), surfaces...) + return seedSystem(ctx, db, input, surfaces...) } -func seedSystem(ctx context.Context, db *gorm.DB, input *system.DatabaseConfig, defaults []platformmodule.TimedTask, surfaces ...platformmodule.Surface) error { +func seedSystem(ctx context.Context, db *gorm.DB, input *system.DatabaseConfig, surfaces ...platformmodule.Surface) error { return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { rootParentID := uint(0) authority := authorityPO{AuthorityID: 888, AuthorityName: "超级管理员", ParentID: &rootParentID, DataScope: 1, DefaultRouter: "dashboard"} @@ -112,18 +111,6 @@ func seedSystem(ctx context.Context, db *gorm.DB, input *system.DatabaseConfig, if err := tx.Where("template_id = ?", exportTemplate.TemplateID).FirstOrCreate(&exportTemplate).Error; err != nil { return err } - var defaultTasks []taskPO - if defaults == nil { - defaultTasks = []taskPO{{Name: "ClearDB", Description: "定时清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Spec: "@daily", ExecutorType: "method", MethodName: "ClearDB", Enabled: true}, {Name: "CleanStaleUploads", Description: "定时清理过期大文件上传会话", Spec: "@hourly", ExecutorType: "method", MethodName: "CleanStaleUploads", Enabled: true}} - } - for _, item := range defaults { - defaultTasks = append(defaultTasks, taskPO{Name: item.Name, Description: item.Description, Spec: item.Spec, WithSeconds: item.WithSeconds, ExecutorType: "method", MethodName: item.MethodName, Enabled: item.Enabled}) - } - for _, task := range defaultTasks { - if err := tx.Where("name = ?", task.Name).FirstOrCreate(&task).Error; err != nil { - return err - } - } for _, item := range input.APIs { if item != nil { po := apiPO{Path: item.Path, Method: strings.ToUpper(item.Method), Description: item.Description, APIGroup: item.APIGroup} diff --git a/internal/data/repository/seed_test.go b/internal/data/system/seed_test.go similarity index 100% rename from internal/data/repository/seed_test.go rename to internal/data/system/seed_test.go diff --git a/internal/data/repository/system_init_ignore_test.go b/internal/data/system/system_init_ignore_test.go similarity index 100% rename from internal/data/repository/system_init_ignore_test.go rename to internal/data/system/system_init_ignore_test.go diff --git a/internal/data/repository/testing_support_test.go b/internal/data/system/testing_support_test.go similarity index 100% rename from internal/data/repository/testing_support_test.go rename to internal/data/system/testing_support_test.go diff --git a/internal/data/repository/transactions_test.go b/internal/data/system/transactions_test.go similarity index 100% rename from internal/data/repository/transactions_test.go rename to internal/data/system/transactions_test.go diff --git a/internal/data/repository/user.go b/internal/data/system/user.go similarity index 100% rename from internal/data/repository/user.go rename to internal/data/system/user.go diff --git a/internal/data/repository/user_strict_test.go b/internal/data/system/user_strict_test.go similarity index 100% rename from internal/data/repository/user_strict_test.go rename to internal/data/system/user_strict_test.go diff --git a/internal/data/repository/version.go b/internal/data/system/version.go similarity index 100% rename from internal/data/repository/version.go rename to internal/data/system/version.go diff --git a/internal/data/repository/version_test.go b/internal/data/system/version_test.go similarity index 100% rename from internal/data/repository/version_test.go rename to internal/data/system/version_test.go diff --git a/internal/data/task/migrations.go b/internal/data/task/migrations.go new file mode 100644 index 0000000..4ae74ee --- /dev/null +++ b/internal/data/task/migrations.go @@ -0,0 +1,46 @@ +package task + +import ( + "context" + "strings" + + "kra/pkg/database/migration" + platformmodule "kra/pkg/module" + + "gorm.io/gorm" +) + +func Migrations() []migration.Step { + return []migration.Step{{ + ID: "202608220001_task_schema", + Migrate: func(db *gorm.DB) error { + return migration.CreateMissingTables(db, &TaskPO{}, &TaskLogPO{}) + }, + }} +} + +// SeedDefaults inserts module-contributed timed tasks without making the +// system data package own task table shapes. +func SeedDefaults(ctx context.Context, db *gorm.DB, defaults []platformmodule.TimedTask) error { + if db == nil { + return nil + } + if len(defaults) == 0 { + defaults = []platformmodule.TimedTask{ + {Name: "ClearDB", Description: "定时清理数据库过期日志", Spec: "@daily", MethodName: "ClearDB", Enabled: true}, + {Name: "CleanStaleUploads", Description: "定时清理过期大文件上传会话", Spec: "@hourly", MethodName: "CleanStaleUploads", Enabled: true}, + } + } + return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + for _, item := range defaults { + if strings.TrimSpace(item.Name) == "" { + continue + } + row := TaskPO{Name: item.Name, Description: item.Description, Spec: item.Spec, WithSeconds: item.WithSeconds, ExecutorType: "method", MethodName: item.MethodName, Enabled: item.Enabled} + if err := tx.Where("name = ?", row.Name).FirstOrCreate(&row).Error; err != nil { + return err + } + } + return nil + }) +} diff --git a/internal/data/task/provider.go b/internal/data/task/provider.go new file mode 100644 index 0000000..c44e68c --- /dev/null +++ b/internal/data/task/provider.go @@ -0,0 +1,10 @@ +package task + +import "gorm.io/gorm" + +// Provider is the minimal database seam required by the task tables. +// Task persistence must not depend on the full data runtime or system repos. +type Provider interface { + DB() *gorm.DB + DatabaseReady() bool +} diff --git a/internal/data/task/providers.go b/internal/data/task/providers.go new file mode 100644 index 0000000..7265f59 --- /dev/null +++ b/internal/data/task/providers.go @@ -0,0 +1,5 @@ +package task + +import "github.com/google/wire" + +var ProviderSet = wire.NewSet(NewTaskRepo) diff --git a/internal/data/repository/task.go b/internal/data/task/task.go similarity index 80% rename from internal/data/repository/task.go rename to internal/data/task/task.go index ef99b8d..0d7d43f 100644 --- a/internal/data/repository/task.go +++ b/internal/data/task/task.go @@ -1,4 +1,4 @@ -package system +package task import ( "context" @@ -10,7 +10,7 @@ import ( "gorm.io/gorm" ) -type taskPO struct { +type TaskPO struct { ID uint `gorm:"primaryKey"` CreatedAt time.Time UpdatedAt time.Time @@ -27,9 +27,9 @@ type taskPO struct { HTTPAllowPrivate, Enabled bool } -func (taskPO) TableName() string { return "sys_timed_tasks" } +func (TaskPO) TableName() string { return "sys_timed_tasks" } -type taskLogPO struct { +type TaskLogPO struct { ID uint `gorm:"primaryKey"` CreatedAt time.Time UpdatedAt time.Time @@ -42,15 +42,15 @@ type taskLogPO struct { ErrorMsg, Output string `gorm:"type:text"` } -func (taskLogPO) TableName() string { return "sys_timed_task_logs" } +func (TaskLogPO) TableName() string { return "sys_timed_task_logs" } type taskRepo struct{ data Provider } func NewTaskRepo(data Provider) taskbiz.TaskRepo { return &taskRepo{data: data} } -func taskToPO(v *taskbiz.TimedTask) taskPO { - return taskPO{ID: v.ID, Name: v.Name, Description: v.Description, Spec: v.Spec, WithSeconds: v.WithSeconds, ExecutorType: v.ExecutorType, MethodName: v.MethodName, Params: gormkit.JSON(v.Params), HTTPURL: v.HTTPURL, HTTPMethod: v.HTTPMethod, HTTPHeader: gormkit.JSON(v.HTTPHeader), HTTPBody: v.HTTPBody, HTTPAllowPrivate: v.HTTPAllowPrivate, Enabled: v.Enabled} +func taskToPO(v *taskbiz.TimedTask) TaskPO { + return TaskPO{ID: v.ID, Name: v.Name, Description: v.Description, Spec: v.Spec, WithSeconds: v.WithSeconds, ExecutorType: v.ExecutorType, MethodName: v.MethodName, Params: gormkit.JSON(v.Params), HTTPURL: v.HTTPURL, HTTPMethod: v.HTTPMethod, HTTPHeader: gormkit.JSON(v.HTTPHeader), HTTPBody: v.HTTPBody, HTTPAllowPrivate: v.HTTPAllowPrivate, Enabled: v.Enabled} } -func taskFromPO(v taskPO) *taskbiz.TimedTask { +func taskFromPO(v TaskPO) *taskbiz.TimedTask { return &taskbiz.TimedTask{ID: v.ID, CreatedAt: v.CreatedAt, UpdatedAt: v.UpdatedAt, Name: v.Name, Description: v.Description, Spec: v.Spec, WithSeconds: v.WithSeconds, ExecutorType: v.ExecutorType, MethodName: v.MethodName, Params: []byte(v.Params), HTTPURL: v.HTTPURL, HTTPMethod: v.HTTPMethod, HTTPHeader: []byte(v.HTTPHeader), HTTPBody: v.HTTPBody, HTTPAllowPrivate: v.HTTPAllowPrivate, Enabled: v.Enabled} } func (r *taskRepo) CreateTask(ctx context.Context, v *taskbiz.TimedTask) error { @@ -63,7 +63,7 @@ func (r *taskRepo) CreateTask(ctx context.Context, v *taskbiz.TimedTask) error { } func (r *taskRepo) TaskNameExists(ctx context.Context, name string, excludeID uint) (bool, error) { var count int64 - db := r.data.DB().WithContext(ctx).Model(&taskPO{}).Where("name = ?", name) + db := r.data.DB().WithContext(ctx).Model(&TaskPO{}).Where("name = ?", name) if excludeID > 0 { db = db.Where("id <> ?", excludeID) } @@ -72,13 +72,13 @@ func (r *taskRepo) TaskNameExists(ctx context.Context, name string, excludeID ui } func (r *taskRepo) UpdateTask(ctx context.Context, v *taskbiz.TimedTask) error { po := taskToPO(v) - return r.data.DB().WithContext(ctx).Model(&taskPO{}).Where("id = ?", v.ID).Select("name", "description", "spec", "with_seconds", "executor_type", "method_name", "params", "http_url", "http_method", "http_header", "http_body", "http_allow_private", "enabled").Updates(&po).Error + return r.data.DB().WithContext(ctx).Model(&TaskPO{}).Where("id = ?", v.ID).Select("name", "description", "spec", "with_seconds", "executor_type", "method_name", "params", "http_url", "http_method", "http_header", "http_body", "http_allow_private", "enabled").Updates(&po).Error } func (r *taskRepo) DeleteTask(ctx context.Context, id uint) error { - return r.data.DB().WithContext(ctx).Delete(&taskPO{}, id).Error + return r.data.DB().WithContext(ctx).Delete(&TaskPO{}, id).Error } func (r *taskRepo) FindTask(ctx context.Context, id uint) (*taskbiz.TimedTask, error) { - var po taskPO + var po TaskPO if err := r.data.DB().WithContext(ctx).First(&po, id).Error; err != nil { return nil, err } @@ -91,7 +91,7 @@ func (r *taskRepo) ListTasks(ctx context.Context, page, size int, q *taskbiz.Tim if !r.data.DatabaseReady() { return []*taskbiz.TimedTask{}, 0, nil } - db := r.data.DB().WithContext(ctx).Model(&taskPO{}) + db := r.data.DB().WithContext(ctx).Model(&TaskPO{}) if q != nil { if q.Name != "" { db = db.Where("name LIKE ?", "%"+q.Name+"%") @@ -107,7 +107,7 @@ func (r *taskRepo) ListTasks(ctx context.Context, page, size int, q *taskbiz.Tim if err := db.Count(&total).Error; err != nil { return nil, 0, err } - var pos []taskPO + var pos []TaskPO if err := pagination.Apply(db.Order("id desc"), page, size, 100).Find(&pos).Error; err != nil { return nil, 0, err } @@ -118,16 +118,16 @@ func (r *taskRepo) ListTasks(ctx context.Context, page, size int, q *taskbiz.Tim return out, total, nil } func (r *taskRepo) ToggleTask(ctx context.Context, id uint, enabled bool) error { - return r.data.DB().WithContext(ctx).Model(&taskPO{}).Where("id = ?", id).Update("enabled", enabled).Error + return r.data.DB().WithContext(ctx).Model(&TaskPO{}).Where("id = ?", id).Update("enabled", enabled).Error } func (r *taskRepo) RecordTaskLog(ctx context.Context, v *taskbiz.TimedTaskLog) error { - return r.data.DB().WithContext(ctx).Create(&taskLogPO{TaskID: v.TaskID, TaskName: v.TaskName, TriggerType: v.TriggerType, StartedAt: v.StartedAt, FinishedAt: v.FinishedAt, DurationMS: v.DurationMS, Status: v.Status, ErrorMsg: v.ErrorMsg, Output: v.Output}).Error + return r.data.DB().WithContext(ctx).Create(&TaskLogPO{TaskID: v.TaskID, TaskName: v.TaskName, TriggerType: v.TriggerType, StartedAt: v.StartedAt, FinishedAt: v.FinishedAt, DurationMS: v.DurationMS, Status: v.Status, ErrorMsg: v.ErrorMsg, Output: v.Output}).Error } -func taskLogFromPO(v taskLogPO) *taskbiz.TimedTaskLog { +func taskLogFromPO(v TaskLogPO) *taskbiz.TimedTaskLog { return &taskbiz.TimedTaskLog{ID: v.ID, CreatedAt: v.CreatedAt, UpdatedAt: v.UpdatedAt, TaskID: v.TaskID, TaskName: v.TaskName, TriggerType: v.TriggerType, StartedAt: v.StartedAt, FinishedAt: v.FinishedAt, DurationMS: v.DurationMS, Status: v.Status, ErrorMsg: v.ErrorMsg, Output: v.Output} } func (r *taskRepo) ListTaskLogs(ctx context.Context, page, size int, taskID uint, status string) ([]*taskbiz.TimedTaskLog, int64, error) { - db := r.data.DB().WithContext(ctx).Model(&taskLogPO{}) + db := r.data.DB().WithContext(ctx).Model(&TaskLogPO{}) if taskID != 0 { db = db.Where("task_id = ?", taskID) } @@ -138,7 +138,7 @@ func (r *taskRepo) ListTaskLogs(ctx context.Context, page, size int, taskID uint if err := db.Count(&total).Error; err != nil { return nil, 0, err } - var pos []taskLogPO + var pos []TaskLogPO if err := pagination.Apply(db.Order("id desc"), page, size, 100).Find(&pos).Error; err != nil { return nil, 0, err } @@ -149,14 +149,7 @@ func (r *taskRepo) ListTaskLogs(ctx context.Context, page, size int, taskID uint return out, total, nil } func (r *taskRepo) CleanupLogs(ctx context.Context) error { - now := time.Now() - return r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { - if err := tx.Unscoped().Where("created_at < ?", now.Add(-2160*time.Hour)).Delete(&operationPO{}).Error; err != nil { - return err - } - if err := tx.Unscoped().Where("created_at < ?", now.Add(-168*time.Hour)).Delete(&jwtBlacklistPO{}).Error; err != nil { - return err - } - return tx.Unscoped().Where("created_at < ?", now.Add(-720*time.Hour)).Delete(&taskLogPO{}).Error - }) + return r.data.DB().WithContext(ctx).Unscoped(). + Where("created_at < ?", time.Now().Add(-720*time.Hour)). + Delete(&TaskLogPO{}).Error } diff --git a/internal/data/repository/task_test.go b/internal/data/task/task_test.go similarity index 95% rename from internal/data/repository/task_test.go rename to internal/data/task/task_test.go index 98faf2c..3f456e6 100644 --- a/internal/data/repository/task_test.go +++ b/internal/data/task/task_test.go @@ -1,4 +1,4 @@ -package system +package task import ( "context" diff --git a/internal/data/task/testing_support_test.go b/internal/data/task/testing_support_test.go new file mode 100644 index 0000000..1220217 --- /dev/null +++ b/internal/data/task/testing_support_test.go @@ -0,0 +1,42 @@ +package task + +import ( + "context" + "fmt" + "sync/atomic" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" +) + +type Data struct { + gormDB *reloadableDB + databaseReady atomic.Bool +} + +func (d *Data) DB() *gorm.DB { + if d == nil || d.gormDB == nil { + return nil + } + return d.gormDB.DB() +} + +func (d *Data) DatabaseReady() bool { return d != nil && d.databaseReady.Load() } + +type reloadableDB struct{ db *gorm.DB } + +func newReloadableDB(db *gorm.DB) *reloadableDB { return &reloadableDB{db: db} } +func (r *reloadableDB) DB() *gorm.DB { + if r == nil { + return nil + } + return r.db +} +func (r *reloadableDB) WithContext(ctx context.Context) *gorm.DB { return r.db.WithContext(ctx) } + +func openWithDriver(driver, dsn string) (*gorm.DB, error) { + if driver != "sqlite" { + return nil, fmt.Errorf("unsupported test database driver %q", driver) + } + return gorm.Open(sqlite.Open(dsn), &gorm.Config{}) +} diff --git a/internal/initialize/initialize.go b/internal/initialize/initialize.go index df0036d..eff2206 100644 --- a/internal/initialize/initialize.go +++ b/internal/initialize/initialize.go @@ -4,7 +4,8 @@ import ( "context" "kra/internal/biz/system" - datasystem "kra/internal/data/repository" + datasystem "kra/internal/data/system" + datatask "kra/internal/data/task" platformmodule "kra/pkg/module" "gorm.io/gorm" @@ -25,7 +26,10 @@ func (r *Repo) IsInitialized(ctx context.Context) (bool, error) { func (r *Repo) Initialize(ctx context.Context, input *system.DatabaseConfig) error { return r.backend.InitializeDatabase(ctx, input, func(ctx context.Context, db *gorm.DB) error { - return datasystem.SeedSystemWithCatalog(ctx, db, input, r.catalog) + if err := datasystem.SeedSystemWithCatalog(ctx, db, input, r.catalog); err != nil { + return err + } + return datatask.SeedDefaults(ctx, db, r.catalog.DefaultTimedTasks()) }) } diff --git a/internal/modules/integration/definition.go b/internal/modules/integration/definition.go new file mode 100644 index 0000000..755c6f6 --- /dev/null +++ b/internal/modules/integration/definition.go @@ -0,0 +1,33 @@ +// Package integration contains the built-in integration data and admin +// surface contribution. +package integration + +import ( + "strings" + + dataintegration "kra/internal/data/integration" + "kra/internal/routecatalog" + "kra/pkg/module" +) + +func Definition() module.Definition { + return module.Definition{ + Name: "integration", + Migrations: dataintegration.Migrations(), + Surface: module.Surface{ + Menus: []module.Menu{{Name: "integrationConfig", Path: "integrationConfig", ParentName: "extensions", Component: "view/systemTools/integration/config.vue", Title: "通信集成", Icon: "connection", Sort: 8}}, + APIs: integrationAPIs(), + }, + } +} + +func integrationAPIs() []module.API { + items := make([]module.API, 0, 5) + for _, descriptor := range routecatalog.Descriptors() { + if descriptor.Public || !strings.HasPrefix(descriptor.Path, "/integration/configs") || descriptor.Description == "" { + continue + } + items = append(items, module.API{Path: descriptor.Path, Method: descriptor.Method, Group: descriptor.Group, Description: descriptor.Description}) + } + return items +} diff --git a/internal/modules/integration/definition_test.go b/internal/modules/integration/definition_test.go new file mode 100644 index 0000000..935dc91 --- /dev/null +++ b/internal/modules/integration/definition_test.go @@ -0,0 +1,10 @@ +package integration + +import "testing" + +func TestDefinitionOwnsCommunicationSurface(t *testing.T) { + definition := Definition() + if definition.Name != "integration" || len(definition.Surface.Menus) != 1 || len(definition.Surface.APIs) != 5 { + t.Fatalf("integration surface = %#v", definition.Surface) + } +} diff --git a/internal/modules/system/definition.go b/internal/modules/system/definition.go index 456c3eb..c1a2a75 100644 --- a/internal/modules/system/definition.go +++ b/internal/modules/system/definition.go @@ -3,38 +3,15 @@ package system import ( - "strings" - - datasystem "kra/internal/data/repository" - "kra/internal/routecatalog" + datasystem "kra/internal/data/system" "kra/pkg/module" ) // Definition describes the built-in system contribution to the application // catalog. Other business modules can expose the same shape independently. func Definition() module.Definition { - communication := module.Surface{ - Menus: []module.Menu{{Name: "integrationConfig", Path: "integrationConfig", ParentName: "extensions", Component: "view/systemTools/integration/config.vue", Title: "通信集成", Icon: "connection", Sort: 8}}, - APIs: integrationAPIs(), - } return module.Definition{ Name: "system", Migrations: datasystem.Migrations(), - Surface: communication, - TimedTasks: []module.TimedTask{ - {Name: "ClearDB", Description: "定时清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Spec: "@daily", MethodName: "ClearDB", Enabled: true}, - {Name: "CleanStaleUploads", Description: "定时清理过期大文件上传会话", Spec: "@hourly", MethodName: "CleanStaleUploads", Enabled: true}, - }, } } - -func integrationAPIs() []module.API { - items := make([]module.API, 0, 5) - for _, descriptor := range routecatalog.Descriptors() { - if descriptor.Public || !strings.HasPrefix(descriptor.Path, "/integration/configs") || descriptor.Description == "" { - continue - } - items = append(items, module.API{Path: descriptor.Path, Method: descriptor.Method, Group: descriptor.Group, Description: descriptor.Description}) - } - return items -} diff --git a/internal/modules/system/definition_test.go b/internal/modules/system/definition_test.go index 52efa38..b64aaf3 100644 --- a/internal/modules/system/definition_test.go +++ b/internal/modules/system/definition_test.go @@ -2,26 +2,9 @@ package system import "testing" -func TestDefinitionIncludesCommunicationIntegrationSurface(t *testing.T) { - surface := Definition().Surface - menuFound := false - for _, menu := range surface.Menus { - if menu.Name == "integrationConfig" { - menuFound = menu.ParentName == "extensions" && menu.Component == "view/systemTools/integration/config.vue" - break - } - } - if !menuFound { - t.Fatal("communication integration menu is missing") - } - apiFound := false - for _, api := range surface.APIs { - if api.Method == "PUT" && api.Path == "/integration/configs/:kind/:provider" { - apiFound = true - break - } - } - if !apiFound { - t.Fatal("communication integration save API is missing") +func TestDefinitionOwnsOnlySystemMigrations(t *testing.T) { + definition := Definition() + if definition.Name != "system" || len(definition.Surface.Menus) != 0 || len(definition.Surface.APIs) != 0 { + t.Fatalf("system definition = %#v", definition) } } diff --git a/internal/modules/task/definition.go b/internal/modules/task/definition.go new file mode 100644 index 0000000..bb91dee --- /dev/null +++ b/internal/modules/task/definition.go @@ -0,0 +1,18 @@ +// Package task contains the built-in timed-task module contribution. +package task + +import ( + datatask "kra/internal/data/task" + "kra/pkg/module" +) + +func Definition() module.Definition { + return module.Definition{ + Name: "task", + Migrations: datatask.Migrations(), + TimedTasks: []module.TimedTask{ + {Name: "ClearDB", Description: "定时清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Spec: "@daily", MethodName: "ClearDB", Enabled: true}, + {Name: "CleanStaleUploads", Description: "定时清理过期大文件上传会话", Spec: "@hourly", MethodName: "CleanStaleUploads", Enabled: true}, + }, + } +} diff --git a/internal/worker/task_registry.go b/internal/worker/task_registry.go index b1bea1c..c702e00 100644 --- a/internal/worker/task_registry.go +++ b/internal/worker/task_registry.go @@ -13,13 +13,14 @@ import ( // TaskMethods is the system module's dependency-bearing task contribution. // Other modules provide their own contributor instead of editing this file. type TaskMethods struct { - tasks *taskbiz.TaskUsecase - media *system.MediaUsecase - runtime *conf.Runtime + tasks *taskbiz.TaskUsecase + maintenance *system.MaintenanceUsecase + media *system.MediaUsecase + runtime *conf.Runtime } -func NewTaskMethods(tasks *taskbiz.TaskUsecase, media *system.MediaUsecase, runtime *conf.Runtime) *TaskMethods { - return &TaskMethods{tasks: tasks, media: media, runtime: runtime} +func NewTaskMethods(tasks *taskbiz.TaskUsecase, maintenance *system.MaintenanceUsecase, media *system.MediaUsecase, runtime *conf.Runtime) *TaskMethods { + return &TaskMethods{tasks: tasks, maintenance: maintenance, media: media, runtime: runtime} } var _ platformtask.Contributor = (*TaskMethods)(nil) @@ -30,6 +31,9 @@ func (methods *TaskMethods) RegisterTasks(registry *platformtask.Registry) { } registry.Register(platformtask.Method{ Name: taskbiz.TaskMethodClearDB, Description: "清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Run: func(ctx context.Context, _ json.RawMessage) error { + if err := methods.maintenance.CleanupExpired(ctx); err != nil { + return err + } return methods.tasks.CleanupLogs(ctx) }, })