yvan/kra-oa
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化结构

Browse Source
This commit is contained in:
Yvan
2026-08-22 21:39:39 +08:00
parent 699bcceeea
commit 0126995090
17 changed files with 61 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/data/README.md
+1 -1
View File
@@ -12,7 +12,7 @@ watching, and repository implementations.
- root files: shared database lifecycle, runtime clients, configuration
orchestration, data-scope auditing, and migration orchestration
新增数据模块时创建 `internal/data/<module>`，提供 `ProviderSet()` 和
新增数据模块时创建 `internal/data/<module>`，提供 `ProviderSet` 和
`Migrations()`，再在 `internal/modules/<module>/definition.go` 注册迁移与
管理面，最后在根 `data.ProviderSet` 中注册该模块。根 `data` 不应直接
拥有业务表 PO，也不应让一个模块引用另一个模块的私有 PO。
internal/data/integration_config.go
-1
View File
@@ -18,7 +18,6 @@ import (
const (
integrationKindStorage = "storage"
integrationKindEmail = "email"
integrationKindPayment = "payment"
)
var storageProviderNames = []string{
internal/data/integration_config_test.go
+4 -2
View File
@@ -7,7 +7,9 @@ import (
"strings"
"testing"
integrationbiz "kra/internal/biz/integration"
"kra/internal/conf"
dataintegration "kra/internal/data/integration"
"kra/internal/integration/storage"
"google.golang.org/protobuf/encoding/protojson"
@@ -73,7 +75,7 @@ func TestStorageIntegrationConfigRoundTrip(t *testing.T) {
if err := saveEmailIntegrationConfig(db, email); err != nil {
t.Fatal(err)
}
if err := db.Create(&dataintegration.ConfigPO{Kind: integrationKindPayment, Provider: "wechat-pay", Config: `{"merchant_id":"123"}`}).Error; err != nil {
if err := db.Create(&dataintegration.ConfigPO{Kind: integrationbiz.IntegrationKindPayment, Provider: "wechat-pay", Config: `{"merchant_id":"123"}`}).Error; err != nil {
t.Fatal(err)
}
@@ -106,7 +108,7 @@ func TestStorageIntegrationConfigRoundTrip(t *testing.T) {
if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindStorage).Count(&storageCount).Error; err != nil {
t.Fatal(err)
}
if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindPayment).Count(&paymentCount).Error; err != nil {
if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationbiz.IntegrationKindPayment).Count(&paymentCount).Error; err != nil {
t.Fatal(err)
}
if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", integrationKindEmail).Count(&emailCount).Error; err != nil {
internal/data/migrations.go
+1 -7
View File
@@ -7,15 +7,9 @@ import (
"gorm.io/gorm"
)
func InfrastructureMigrations() []migration.Step {
return nil
}
// 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)
return migration.Run(db, catalog.MigrationSteps())
}
internal/data/migrations_test.go
+1
View File
@@ -3,6 +3,7 @@ package data
import (
"testing"
dataintegration "kra/internal/data/integration"
integrationmodule "kra/internal/modules/integration"
paymentmodule "kra/internal/modules/payment"
systemmodule "kra/internal/modules/system"
internal/data/payment/models.go
-5
View File
@@ -1,5 +0,0 @@
package payment
import dataintegration "kra/internal/data/integration"
const integrationKindPayment = "payment"
internal/data/payment/payment.go
+6 -5
View File
@@ -9,6 +9,7 @@ import (
"fmt"
integrationbiz "kra/internal/biz/integration"
bizpayment "kra/internal/biz/payment"
dataintegration "kra/internal/data/integration"
"net/url"
"strconv"
"strings"
@@ -27,11 +28,11 @@ func NewPaymentRepo(data Provider) bizpayment.PaymentRepo { return &paymentRepo{
func ensurePaymentIntegrationConfigs(db *gorm.DB) error {
for _, provider := range bizpayment.SupportedPaymentProviders {
var row dataintegration.ConfigPO
err := db.Where("kind = ? AND provider = ?", integrationKindPayment, provider).First(&row).Error
defaults := integrationbiz.DefaultIntegrationConfig(integrationKindPayment, provider)
err := db.Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error
defaults := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindPayment, provider)
if errors.Is(err, gorm.ErrRecordNotFound) {
encoded, _ := json.Marshal(defaults)
if err := db.Create(&dataintegration.ConfigPO{Kind: integrationKindPayment, Provider: provider, Enabled: false, Config: string(encoded)}).Error; err != nil {
if err := db.Create(&dataintegration.ConfigPO{Kind: integrationbiz.IntegrationKindPayment, Provider: provider, Enabled: false, Config: string(encoded)}).Error; err != nil {
return err
}
continue
@@ -60,7 +61,7 @@ func ensurePaymentIntegrationConfigs(db *gorm.DB) error {
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 err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil, bizpayment.ErrPaymentProviderNotFound
}
@@ -299,7 +300,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 dataintegration.ConfigPO
if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationKindPayment, provider).First(&row).Error; err != nil {
if err := r.data.DB().WithContext(ctx).Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindPayment, provider).First(&row).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, bizpayment.ErrPaymentProviderNotFound
}
internal/data/payment/testing_support_test.go
+1
View File
@@ -6,6 +6,7 @@ import (
"github.com/glebarez/sqlite"
"gorm.io/gorm"
dataintegration "kra/internal/data/integration"
)
type Data struct{ gormDB *reloadableDB }
internal/data/task/migrations.go
+2 -2
View File
@@ -14,7 +14,7 @@ func Migrations() []migration.Step {
return []migration.Step{{
ID: "202608220001_task_schema",
Migrate: func(db *gorm.DB) error {
return migration.CreateMissingTables(db, &TaskPO{}, &TaskLogPO{})
return migration.CreateMissingTables(db, &taskPO{}, &taskLogPO{})
},
}}
}
@@ -36,7 +36,7 @@ func SeedDefaults(ctx context.Context, db *gorm.DB, defaults []platformmodule.Ti
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}
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
}
internal/data/task/task.go
+20 -20
View File
@@ -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
}
@@ -148,8 +148,8 @@ func (r *taskRepo) ListTaskLogs(ctx context.Context, page, size int, taskID uint
}
return out, total, nil
}
func (r *taskRepo) CleanupLogs(ctx context.Context) error {
func (r *taskRepo) CleanupTaskLogs(ctx context.Context) error {
return r.data.DB().WithContext(ctx).Unscoped().
Where("created_at < ?", time.Now().Add(-720*time.Hour)).
Delete(&TaskLogPO{}).Error
Delete(&taskLogPO{}).Error
}