优化结构

2026-08-21 00:01:48 +08:00
parent e65cf0b1d0
commit 746e33e027
344 changed files with 1021 additions and 578 deletions
@@ -0,0 +1,13 @@
+# Platform
+
+这里放跨业务模块的基础设施，不放具体业务表或业务用例。
+
+当前包含：
+
+- `database`：跨模块共享的 GORM 支持、分页和迁移执行器
+- `module`：模块迁移、后台元数据和路由注册协议
+- `task`：跨模块共享的进程内任务注册表和贡献协议
+
+平台层只能提供机制，不能引用 `internal/modules/*`。数据库配置加载、系统集成配置
+和系统表仍由 system 模块负责；未来提炼公共连接管理时，也必须先形成不带系统
+业务语义的稳定接口，再移动到这里。
@@ -0,0 +1,5 @@
+# GORM Helpers
+
+This package owns storage-specific GORM types and logging integration used by
+the data repositories. It is intentionally internal because its JSON value and
+logger are persistence concerns, not general application utilities.
@@ -0,0 +1,105 @@
+// Package gormkit contains reusable GORM integration helpers.
+package gormkit
+
+import (
+	"context"
+	"database/sql/driver"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+	"gorm.io/gorm/schema"
+)
+
+// JSON is a portable JSON value for GORM models. It selects a native JSON
+// column where the driver supports one and a textual fallback elsewhere.
+type JSON []byte
+
+func (JSON) GormDataType() string { return "json" }
+func (JSON) GormDBDataType(db *gorm.DB, _ *schema.Field) string {
+	switch db.Dialector.Name() {
+	case "mysql", "sqlite":
+		return "JSON"
+	case "postgres":
+		return "JSONB"
+	case "sqlserver":
+		return "NVARCHAR(MAX)"
+	case "oracle":
+		return "CLOB"
+	default:
+		return "TEXT"
+	}
+}
+func (j JSON) Value() (driver.Value, error) {
+	if len(j) == 0 {
+		return nil, nil
+	}
+	return string(j), nil
+}
+func (j *JSON) Scan(value any) error {
+	switch raw := value.(type) {
+	case nil:
+		*j = nil
+	case []byte:
+		*j = append((*j)[:0], raw...)
+	case string:
+		*j = append((*j)[:0], raw...)
+	default:
+		return fmt.Errorf("cannot scan JSON from %T", value)
+	}
+	return nil
+}
+
+// Logger forwards GORM diagnostics through an application slog logger.
+type Logger struct {
+	logger        *slog.Logger
+	slowThreshold time.Duration
+	level         logger.LogLevel
+}
+
+func NewLogger(log *slog.Logger, level logger.LogLevel) *Logger {
+	if log == nil {
+		log = slog.New(slog.NewTextHandler(io.Discard, nil))
+	}
+	return &Logger{logger: log, slowThreshold: 200 * time.Millisecond, level: level}
+}
+
+func (g *Logger) LogMode(level logger.LogLevel) logger.Interface {
+	next := *g
+	next.level = level
+	return &next
+}
+
+func (g *Logger) Info(ctx context.Context, message string, args ...any) {
+	g.logger.InfoContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
+}
+
+func (g *Logger) Warn(ctx context.Context, message string, args ...any) {
+	g.logger.WarnContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
+}
+
+func (g *Logger) Error(ctx context.Context, message string, args ...any) {
+	g.logger.ErrorContext(ctx, fmt.Sprintf(message, args...), "mod", "sql", "gorm_logger", true)
+}
+
+func (g *Logger) Trace(ctx context.Context, begin time.Time, query func() (string, int64), queryErr error) {
+	if g.level <= logger.Silent {
+		return
+	}
+	elapsed := time.Since(begin)
+	sql, rows := query()
+	fields := []any{"mod", "sql", "gorm_logger", true, "sql", sql, "rows", rows, "elapsed_ms", elapsed.Milliseconds()}
+	switch {
+	case queryErr != nil && g.level >= logger.Error && !errors.Is(queryErr, logger.ErrRecordNotFound):
+		fields = append(fields, "error", queryErr)
+		g.logger.ErrorContext(ctx, "SQL 执行错误", fields...)
+	case elapsed > g.slowThreshold && g.level >= logger.Warn:
+		g.logger.WarnContext(ctx, "SQL 慢查询", fields...)
+	case g.level >= logger.Info:
+		g.logger.InfoContext(ctx, "SQL", fields...)
+	}
+}
@@ -0,0 +1,121 @@
+// Package migration owns the application database migration runner. Each data
+// module declares its own steps; the root data package only orders and runs
+// them.
+package migration
+
+import (
+	"fmt"
+	"strings"
+
+	"gorm.io/gorm"
+)
+
+const TableName = "sys_schema_migrations"
+
+type Step struct {
+	ID      string
+	Migrate func(*gorm.DB) error
+}
+
+type record struct {
+	ID string `gorm:"primaryKey;size:255"`
+}
+
+func (record) TableName() string { return TableName }
+
+func Run(db *gorm.DB, steps []Step) error {
+	if db == nil {
+		return fmt.Errorf("database is nil")
+	}
+	db = db.Session(&gorm.Session{NewDB: true})
+	if err := repairMigrationTable(db); err != nil {
+		return err
+	}
+	if !db.Session(&gorm.Session{NewDB: true}).Migrator().HasTable(&record{}) {
+		if err := db.Session(&gorm.Session{NewDB: true}).Migrator().CreateTable(&record{}); err != nil {
+			return fmt.Errorf("create migration table: %w", err)
+		}
+	}
+	var appliedRows []record
+	if err := db.Session(&gorm.Session{NewDB: true}).Find(&appliedRows).Error; err != nil {
+		return fmt.Errorf("load applied migrations: %w", err)
+	}
+	applied := make(map[string]struct{}, len(appliedRows))
+	known := make(map[string]struct{}, len(steps))
+	for _, row := range appliedRows {
+		applied[row.ID] = struct{}{}
+	}
+	for _, step := range steps {
+		if step.ID == "" {
+			return fmt.Errorf("migration id is empty")
+		}
+		if _, duplicate := known[step.ID]; duplicate {
+			return fmt.Errorf("duplicate migration id %q", step.ID)
+		}
+		known[step.ID] = struct{}{}
+	}
+	for id := range applied {
+		if _, ok := known[id]; !ok {
+			return fmt.Errorf("unknown applied migration %q", id)
+		}
+	}
+	for _, step := range steps {
+		if _, ok := applied[step.ID]; ok {
+			continue
+		}
+		if step.Migrate == nil {
+			return fmt.Errorf("migration %q has no function", step.ID)
+		}
+		if err := step.Migrate(db.Session(&gorm.Session{NewDB: true})); err != nil {
+			return fmt.Errorf("apply migration %q: %w", step.ID, err)
+		}
+		if err := db.Session(&gorm.Session{NewDB: true}).Create(&record{ID: step.ID}).Error; err != nil {
+			return fmt.Errorf("record migration %q: %w", step.ID, err)
+		}
+	}
+	return nil
+}
+
+func repairMigrationTable(db *gorm.DB) error {
+	clean := db.Session(&gorm.Session{NewDB: true})
+	if !clean.Migrator().HasTable(TableName) {
+		return nil
+	}
+	var dataType string
+	switch clean.Dialector.Name() {
+	case "mysql":
+		err := clean.Raw("SELECT DATA_TYPE FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?", TableName, "id").Scan(&dataType).Error
+		if err != nil {
+			return fmt.Errorf("inspect migration table: %w", err)
+		}
+	case "postgres":
+		err := clean.Raw("SELECT data_type FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ? AND column_name = ?", TableName, "id").Scan(&dataType).Error
+		if err != nil {
+			return fmt.Errorf("inspect migration table: %w", err)
+		}
+	case "sqlite":
+		var rows []struct {
+			Name string `gorm:"column:name"`
+			Type string `gorm:"column:type"`
+		}
+		if err := clean.Raw("PRAGMA table_info(" + TableName + ")").Scan(&rows).Error; err != nil {
+			return fmt.Errorf("inspect migration table: %w", err)
+		}
+		for _, row := range rows {
+			if strings.EqualFold(row.Name, "id") {
+				dataType = row.Type
+				break
+			}
+		}
+	default:
+		return nil
+	}
+	normalized := strings.ToLower(dataType)
+	if strings.Contains(normalized, "char") || strings.Contains(normalized, "text") || strings.Contains(normalized, "clob") {
+		return nil
+	}
+	if err := clean.Migrator().DropTable(TableName); err != nil {
+		return fmt.Errorf("repair invalid migration table: %w", err)
+	}
+	return nil
+}
@@ -0,0 +1,20 @@
+package migration
+
+import "gorm.io/gorm"
+
+// CreateMissingTables creates only schemas that are absent. Existing tables
+// are intentionally left untouched: GORM's MySQL column diff path can panic
+// when older servers return incomplete information_schema metadata, and this
+// project does not support implicit legacy-schema upgrades through AutoMigrate.
+func CreateMissingTables(db *gorm.DB, models ...any) error {
+	missing := make([]any, 0, len(models))
+	for _, model := range models {
+		if model != nil && !db.Migrator().HasTable(model) {
+			missing = append(missing, model)
+		}
+	}
+	if len(missing) == 0 {
+		return nil
+	}
+	return db.Migrator().CreateTable(missing...)
+}
@@ -0,0 +1,104 @@
+package migration
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+)
+
+type existingSchema struct {
+	ID   uint `gorm:"primaryKey"`
+	Name string
+}
+
+func (existingSchema) TableName() string { return "existing_schema" }
+
+func TestCreateMissingTablesDoesNotAlterExistingTables(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Exec("CREATE TABLE existing_schema (id integer primary key)").Error; err != nil {
+		t.Fatal(err)
+	}
+	if err = CreateMissingTables(db, &existingSchema{}); err != nil {
+		t.Fatal(err)
+	}
+	if db.Migrator().HasColumn(&existingSchema{}, "name") {
+		t.Fatal("existing schema was altered")
+	}
+}
+
+func TestCreateMissingTablesCreatesAbsentTables(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = CreateMissingTables(db, &existingSchema{}); err != nil {
+		t.Fatal(err)
+	}
+	if !db.Migrator().HasTable(&existingSchema{}) || !db.Migrator().HasColumn(&existingSchema{}, "name") {
+		t.Fatal("missing schema was not created")
+	}
+}
+
+func TestRunRepairsInvalidMigrationMetadataTable(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Exec("CREATE TABLE sys_schema_migrations (id integer primary key, title text)").Error; err != nil {
+		t.Fatal(err)
+	}
+	called := false
+	if err = Run(db, []Step{{ID: "202608200001_test", Migrate: func(*gorm.DB) error {
+		called = true
+		return nil
+	}}}); err != nil {
+		t.Fatal(err)
+	}
+	if !called {
+		t.Fatal("migration was not rerun after metadata repair")
+	}
+	var dataType string
+	if err = db.Raw("SELECT type FROM pragma_table_info(?) WHERE name = 'id'", TableName).Scan(&dataType).Error; err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(strings.ToLower(dataType), "text") {
+		t.Fatalf("migration id type = %q", dataType)
+	}
+}
+
+func TestRunDoesNotLeakStatementState(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = Run(db, []Step{{ID: "202608200001_test", Migrate: func(tx *gorm.DB) error {
+		return CreateMissingTables(tx, &existingSchema{})
+	}}}); err != nil {
+		t.Fatal(err)
+	}
+	var count int64
+	if err = db.Model(&existingSchema{}).Count(&count).Error; err != nil {
+		t.Fatalf("shared database retained migration statement state: %v", err)
+	}
+}
+
+func TestRunRejectsUnknownAppliedMigration(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Migrator().CreateTable(&record{}); err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Create(&record{ID: "unknown"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err = Run(db, []Step{{ID: "known", Migrate: func(*gorm.DB) error { return nil }}}); err == nil {
+		t.Fatal("unknown migration was accepted")
+	}
+}
@@ -0,0 +1,37 @@
+// Package pagination contains GORM pagination helpers shared by repository
+// implementations in the data layer.
+package pagination
+
+import "gorm.io/gorm"
+
+// LimitOffset normalizes a page request and applies the optional maximum page
+// size. A non-positive size means that no limit should be applied.
+func LimitOffset(page, size, maxSize int) (limit, offset int) {
+	limit = size
+	if maxSize > 0 && size > maxSize {
+		limit = maxSize
+	}
+	if limit <= 0 {
+		return 0, 0
+	}
+	if page <= 0 {
+		page = 1
+	}
+	return limit, (page - 1) * limit
+}
+
+// Apply applies pagination only when the caller supplied a positive size.
+func Apply(db *gorm.DB, page, size, maxSize int) *gorm.DB {
+	limit, offset := LimitOffset(page, size, maxSize)
+	if limit == 0 {
+		return db
+	}
+	return db.Offset(offset).Limit(limit)
+}
+
+// ApplyRequired always applies LIMIT/OFFSET, including LIMIT 0. This keeps
+// the existing repository contract for APIs whose page size is mandatory.
+func ApplyRequired(db *gorm.DB, page, size, maxSize int) *gorm.DB {
+	limit, offset := LimitOffset(page, size, maxSize)
+	return db.Offset(offset).Limit(limit)
+}
@@ -0,0 +1,102 @@
+// Package module defines the small contracts shared by application modules.
+// A module contributes its own schema, seed data, admin surface and runtime
+// hooks without depending on the built-in system implementation.
+package module
+
+import (
+	"github.com/gin-gonic/gin"
+	"kra/pkg/database/migration"
+	"kra/pkg/task"
+)
+
+type Menu struct {
+	Name, Path, ParentName, Component, Title, Icon string
+	Sort                                           int
+}
+
+type API struct {
+	Path, Method, Group, Description string
+}
+
+type Surface struct {
+	Menus []Menu
+	APIs  []API
+}
+
+type TimedTask struct {
+	Name, Description, Spec, MethodName string
+	WithSeconds, Enabled                bool
+}
+
+type Definition struct {
+	Name       string
+	Migrations []migration.Step
+	Surface    Surface
+	TimedTasks []TimedTask
+	Tasks      []task.Method
+}
+
+type Catalog struct {
+	Definitions []Definition
+}
+
+func (c Catalog) MigrationSteps() []migration.Step {
+	var steps []migration.Step
+	for _, item := range c.Definitions {
+		steps = append(steps, item.Migrations...)
+	}
+	return steps
+}
+
+func (c Catalog) Surface() Surface {
+	var surface Surface
+	for _, item := range c.Definitions {
+		surface.Menus = append(surface.Menus, item.Surface.Menus...)
+		surface.APIs = append(surface.APIs, item.Surface.APIs...)
+	}
+	return surface
+}
+
+func (c Catalog) DefaultTimedTasks() []TimedTask {
+	var tasks []TimedTask
+	for _, item := range c.Definitions {
+		tasks = append(tasks, item.TimedTasks...)
+	}
+	return tasks
+}
+
+func (c Catalog) TaskMethods() []task.Method {
+	var methods []task.Method
+	for _, item := range c.Definitions {
+		methods = append(methods, item.Tasks...)
+	}
+	return methods
+}
+
+type RouteRegistrar interface {
+	RegisterRoutes(public, private *gin.RouterGroup, engine *gin.Engine)
+}
+
+type Runtime struct {
+	routes []RouteRegistrar
+}
+
+func NewRuntime(routes ...RouteRegistrar) *Runtime {
+	return &Runtime{routes: append([]RouteRegistrar(nil), routes...)}
+}
+
+func (r *Runtime) Add(routes ...RouteRegistrar) {
+	if r == nil {
+		return
+	}
+	r.routes = append(r.routes, routes...)
+}
+
+func (r *Runtime) RegisterRoutes(public, private *gin.RouterGroup, engine *gin.Engine) {
+	if r == nil {
+		return
+	}
+	for _, registrar := range r.routes {
+		registrar.RegisterRoutes(public, private, engine)
+	}
+}
@@ -0,0 +1,43 @@
+package module
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"kra/pkg/database/migration"
+	"kra/pkg/task"
+)
+
+func TestCatalogCollectsContributions(t *testing.T) {
+	run := func(context.Context, json.RawMessage) error { return nil }
+	catalog := Catalog{Definitions: []Definition{
+		{
+			Name:       "system",
+			Migrations: []migration.Step{{ID: "system_schema"}},
+			Surface:    Surface{Menus: []Menu{{Name: "users"}}, APIs: []API{{Path: "/users"}}},
+			TimedTasks: []TimedTask{{Name: "system.cleanup"}},
+			Tasks:      []task.Method{{Name: "system.cleanup", Run: run}},
+		},
+		{
+			Name:       "orders",
+			Migrations: []migration.Step{{ID: "orders_schema"}},
+			Surface:    Surface{Menus: []Menu{{Name: "orders"}}, APIs: []API{{Path: "/orders"}}},
+			TimedTasks: []TimedTask{{Name: "orders.expire"}},
+			Tasks:      []task.Method{{Name: "orders.expire", Run: run}},
+		},
+	}}
+
+	if got := catalog.MigrationSteps(); len(got) != 2 || got[0].ID != "system_schema" || got[1].ID != "orders_schema" {
+		t.Fatalf("unexpected migrations: %#v", got)
+	}
+	if got := catalog.Surface(); len(got.Menus) != 2 || len(got.APIs) != 2 || got.Menus[1].Name != "orders" {
+		t.Fatalf("unexpected surface: %#v", got)
+	}
+	if got := catalog.DefaultTimedTasks(); len(got) != 2 || got[1].Name != "orders.expire" {
+		t.Fatalf("unexpected default tasks: %#v", got)
+	}
+	if got := catalog.TaskMethods(); len(got) != 2 || got[1].Name != "orders.expire" {
+		t.Fatalf("unexpected task methods: %#v", got)
+	}
+}
@@ -0,0 +1,5 @@
+package task
+
+import "github.com/google/wire"
+
+var ProviderSet = wire.NewSet(NewRegistry)
@@ -0,0 +1,88 @@
+// Package task contains the process-wide task method contract. It is an
+// infrastructure registry, not a system business package, so any module can
+// contribute an in-process scheduled task.
+package task
+
+import (
+	"context"
+	"encoding/json"
+	"sort"
+	"sync"
+)
+
+type MethodFunc func(context.Context, json.RawMessage) error
+
+type Method struct {
+	Name, Description string
+	Run               MethodFunc
+}
+
+// Contributor registers task methods whose handlers need constructed runtime
+// dependencies. Modules expose a contributor and the application composition
+// root activates it; the task runtime never imports business modules.
+type Contributor interface {
+	RegisterTasks(*Registry)
+}
+
+type Registry struct {
+	mu      sync.RWMutex
+	methods map[string]Method
+}
+
+func NewRegistry() *Registry { return &Registry{methods: make(map[string]Method)} }
+
+func Apply(registry *Registry, contributors ...Contributor) {
+	if registry == nil {
+		return
+	}
+	for _, contributor := range contributors {
+		if contributor != nil {
+			contributor.RegisterTasks(registry)
+		}
+	}
+}
+
+func (r *Registry) Register(method Method) {
+	if r == nil || method.Name == "" || method.Run == nil {
+		return
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.methods == nil {
+		r.methods = make(map[string]Method)
+	}
+	r.methods[method.Name] = method
+}
+
+func (r *Registry) RegisterAll(methods []Method) {
+	for _, method := range methods {
+		r.Register(method)
+	}
+}
+
+func (r *Registry) Lookup(name string) (MethodFunc, bool) {
+	if r == nil {
+		return nil, false
+	}
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	method, ok := r.methods[name]
+	if !ok {
+		return nil, false
+	}
+	return method.Run, true
+}
+
+func (r *Registry) List() []Method {
+	if r == nil {
+		return nil
+	}
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	methods := make([]Method, 0, len(r.methods))
+	for _, method := range r.methods {
+		methods = append(methods, method)
+	}
+	sort.Slice(methods, func(i, j int) bool { return methods[i].Name < methods[j].Name })
+	return methods
+}
@@ -0,0 +1,32 @@
+package task
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+)
+
+func TestRegistryIsComposableAndIdempotent(t *testing.T) {
+	registry := NewRegistry()
+	registry.RegisterAll([]Method{{Name: "orders.sync", Run: func(context.Context, json.RawMessage) error { return nil }}, {Name: "system.cleanup", Run: func(context.Context, json.RawMessage) error { return nil }}})
+	registry.Register(Method{Name: "orders.sync", Description: "updated", Run: func(context.Context, json.RawMessage) error { return nil }})
+	methods := registry.List()
+	if len(methods) != 2 || methods[0].Name != "orders.sync" || methods[0].Description != "updated" {
+		t.Fatalf("unexpected methods: %#v", methods)
+	}
+}
+
+type testContributor struct{ name string }
+
+func (contributor testContributor) RegisterTasks(registry *Registry) {
+	registry.Register(Method{Name: contributor.name, Run: func(context.Context, json.RawMessage) error { return nil }})
+}
+
+func TestApplyCollectsModuleContributors(t *testing.T) {
+	registry := NewRegistry()
+	Apply(registry, testContributor{name: "system.cleanup"}, testContributor{name: "orders.expire"})
+	methods := registry.List()
+	if len(methods) != 2 || methods[0].Name != "orders.expire" || methods[1].Name != "system.cleanup" {
+		t.Fatalf("unexpected contributed methods: %#v", methods)
+	}
+}