diff --git a/.gitignore b/.gitignore index 547e197..b979b7a 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ bin/ dist/ *.log coverage.out +__pycache__/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 930ae3f..8046069 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## v0.2.0 + +- 新增 `go-omni new --database mysql`,生成数据库配置、MySQL Repository、ServiceContext、`schema.sql` 和数据库就绪探针;默认仍使用内存存储。 +- 两种存储模式的脚手架均提供完整 CRUD、分页、输入校验与统一响应。新生成项目的 `POST /users` 从直接返回用户对象改为标准响应封装,用户对象位于 `data`。 +- MySQL 启动时检查连接和表结构,失败时清理连接池;数据库停机后 readiness 返回 503,恢复后自动重新检查。 +- MySQL Repository 支持 `*sql.DB` 和 `*sql.Tx`,可在同一事务中复用业务查询。 +- 修复 `MaxIdleConns=0` 未生效的问题,数据库健康检查增加超时限制。 +- 新增隔离 MySQL 容器验收脚本,覆盖真实 CRUD、事务提交和错误/panic/取消回滚、持久化、资源清理及数据库停机恢复。 + ## v0.1.1 - 修复脚手架生成项目固定依赖不存在的 `v0.0.0`;远程安装的 CLI 默认使用自身版本,支持 `--framework-version` 指定完整版本号。 diff --git a/README.md b/README.md index e58b28c..b277bee 100644 --- a/README.md +++ b/README.md @@ -5,13 +5,13 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认用 Gi 远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本: ```sh -go get git.yuntuo.hk/k2safe/go-omni@v0.1.1 +go get git.yuntuo.hk/k2safe/go-omni@v0.2.0 ``` 创建新业务项目: ```sh -go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.1.1 +go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.2.0 go-omni new --module example.com/billing ./billing cd billing go mod tidy @@ -21,6 +21,10 @@ go run ./cmd/server 远程安装的 CLI 默认引用自身版本,生成的项目不需要本地 `replace`。 具体版本和本地开发选项见 [CLI 文档](cmd/go-omni/README.md)。 +需要数据库持久化时,使用 `go-omni new --database mysql --module example.com/billing ./billing`。 +生成项目包含完整用户 CRUD、数据库 Repository、ServiceContext、建表脚本和就绪探针。 +先应用生成的 `schema.sql`,再配置 `GO_OMNI_DB_DSN`;启动时会检查连接与表结构,退出时释放连接池。 + ## 目录结构 ```text @@ -148,7 +152,11 @@ application.GET("/health/ready", checker.ReadinessHandler()) 新项目脚手架会生成项目自己的 `internal/servicecontext`。它集中保存 Logger、数据库、缓存和外部客户端,Service 只依赖这个项目上下文,不把基础设施初始化散落到各个 Handler: ```go -services := servicecontext.New(logger, users) +services, err := servicecontext.New(ctx, logger) +if err != nil { + return err +} +defer services.Close() userService := service.NewUsers(services) ``` diff --git a/ROADMAP.md b/ROADMAP.md index 58ba3e8..0ae058c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -8,11 +8,15 @@ - 自动验收入口:`python3 scripts/verifyScaffold.py --version v0.1.1`。 - 已通过全量 race 测试、vet、build,以及从远程提交安装 CLI 后的无 replace 端到端验收;覆盖生成服务和 `App.Run` 的两种退出信号。 -## 2. 业务数据接入(待完成) +## 2. 业务数据接入(已完成,v0.2.0) - 数据库初始化、Repository 和 ServiceContext 组合成完整业务启动流程。 - Readiness 注册数据库探针,启动失败与服务退出时清理连接池。 - 验收:真实数据库 CRUD、事务回滚、数据库不可用时 readiness 返回 503。 +- `go-omni new --database mysql` 已生成完整数据接入流程,默认内存模式仍然可用。 +- 独立 MySQL 8.4 容器已验证 CRUD、唯一邮箱、分页、服务重启持久化、事务提交与错误/panic/取消回滚。 +- 已验证初始化失败、缺少表结构、正常退出时关闭连接池,以及数据库停止后的 readiness 503、恢复后的 200。 +- 自动验收入口:`python3 scripts/verifyDatabase.py --version v0.2.0`;结束后清理本次测试容器、匿名卷和临时项目。 ## 3. 网络适配一致性(待完成) @@ -25,4 +29,4 @@ - 在一个实际业务模块中使用远程固定版本,验证配置、日志、错误处理和升级流程。 - 根据真实接入问题收敛公共 API,再评估稳定版范围。 -以上阶段以验收结果判断完成。当前仍是基础框架试用阶段,数据库生产验收、自研网络内核与实际业务接入尚未完成。 +以上阶段以验收结果判断完成。当前仍是基础框架试用阶段;MySQL 功能验收已完成,生产压测、自研网络内核与实际业务接入尚未完成。 diff --git a/cmd/go-omni/README.md b/cmd/go-omni/README.md index 49decf9..2067563 100644 --- a/cmd/go-omni/README.md +++ b/cmd/go-omni/README.md @@ -3,7 +3,7 @@ 从远程安装,生成一个带 Handler / Service / Repository / Model 分层的服务骨架: ```sh -go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.1.1 +go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.2.0 go-omni new --module example.com/billing ./billing cd billing go mod tidy @@ -16,7 +16,7 @@ go run ./cmd/server 需要选择其他已发布版本时,使用完整版本号: ```sh -go-omni new --framework-version v0.1.1 --module example.com/billing ./billing +go-omni new --framework-version v0.2.0 --module example.com/billing ./billing ``` `--framework-version` 禁用自动本地替换;`main`、`latest` 等浮动引用不接受为固定版本。 @@ -32,13 +32,24 @@ go-omni new --framework-version v0.1.1 --module example.com/billing ./billing go-omni new --framework-dir /path/to/go-omni --module example.com/billing ./billing ``` -生成服务默认包含 `/health/live`、`/health/ready` 和示例 `POST /users`。 +默认生成内存存储版本;选择 MySQL 时: + +```sh +go-omni new --database mysql --module example.com/billing ./billing +``` + +MySQL 项目额外包含 `schema.sql`、MySQL Repository 和可选的数据库集成测试。 +先应用建表脚本,再设置 `GO_OMNI_DB_DSN` 启动。ServiceContext 负责连接池初始化、表结构检查、 +就绪探针注册和关闭。两种模式都提供用户 CRUD、分页、输入校验和统一错误响应。 + +生成服务默认包含 `/health/live`、`/health/ready` 和示例 `/users` CRUD。 退出信号通过 `signal.NotifyContext` 传入 `RunContext`,Logger 的关闭函数注册到应用生命周期。 发布验收使用 Python 3 标准库脚本,不需要额外 Python 依赖: ```sh -python3 scripts/verifyScaffold.py --version v0.1.1 +python3 scripts/verifyScaffold.py --version v0.2.0 +python3 scripts/verifyDatabase.py --version v0.2.0 ``` 脚本在源码目录外安装远程 CLI、生成项目,确认固定版本且没有 `replace`,再下载依赖、 @@ -50,4 +61,7 @@ python3 scripts/verifyScaffold.py --version v0.1.1 ```sh go build -o /tmp/go-omni ./cmd/go-omni python3 scripts/verifyScaffold.py --cli /tmp/go-omni --framework-dir . +python3 scripts/verifyDatabase.py --cli /tmp/go-omni --framework-dir . ``` + +数据库验收需要本地 Docker,使用独立 MySQL 8.4 容器;不会连接已有业务数据库,结束后清理测试容器与数据卷。 diff --git a/cmd/go-omni/main.go b/cmd/go-omni/main.go index a0914f6..7a60227 100644 --- a/cmd/go-omni/main.go +++ b/cmd/go-omni/main.go @@ -39,6 +39,17 @@ func runNew(args []string) error { switch { case arg == "--force": options.Force = true + case arg == "--database": + if index+1 >= len(args) || strings.TrimSpace(args[index+1]) == "" { + return errors.New("--database requires memory or mysql") + } + index++ + options.Database = args[index] + case strings.HasPrefix(arg, "--database="): + options.Database = strings.TrimPrefix(arg, "--database=") + if strings.TrimSpace(options.Database) == "" { + return errors.New("--database requires memory or mysql") + } case arg == "--module": if index+1 >= len(args) { return errors.New("--module requires a value") @@ -95,4 +106,5 @@ func usage() { fmt.Fprintln(os.Stderr, " go-omni new [--module ] [--force] ") fmt.Fprintln(os.Stderr, " --framework-version Pin a remote framework version (defaults to CLI version)") fmt.Fprintln(os.Stderr, " --framework-dir Use a local framework checkout") + fmt.Fprintln(os.Stderr, " --database Select generated storage (default memory)") } diff --git a/cmd/go-omni/main_test.go b/cmd/go-omni/main_test.go index 37e9087..64f56df 100644 --- a/cmd/go-omni/main_test.go +++ b/cmd/go-omni/main_test.go @@ -31,3 +31,21 @@ func TestNewRejectsEmptyFrameworkVersion(t *testing.T) { } } } + +func TestNewMySQLDatabaseOptions(t *testing.T) { + for _, flag := range [][]string{{"--database", "mysql"}, {"--database=mysql"}} { + project := filepath.Join(t.TempDir(), "billing") + args := append([]string{"--module", "example.com/billing"}, flag...) + if err := runNew(append(args, project)); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(project, "schema.sql")); err != nil { + t.Fatal(err) + } + } + for _, args := range [][]string{{"--database"}, {"--database="}, {"--database", " "}} { + if err := runNew(args); err == nil { + t.Fatalf("accepted missing database type: %v", args) + } + } +} diff --git a/cmd/go-omni/scaffold/scaffold.go b/cmd/go-omni/scaffold/scaffold.go index 4023b7f..da903b4 100644 --- a/cmd/go-omni/scaffold/scaffold.go +++ b/cmd/go-omni/scaffold/scaffold.go @@ -4,6 +4,7 @@ import ( "embed" "errors" "fmt" + "go/format" "io/fs" "os" "path/filepath" @@ -22,6 +23,7 @@ type Options struct { Module string FrameworkDir string FrameworkVersion string + Database string Force bool } @@ -30,6 +32,8 @@ var files = map[string]string{ "README.md": "templates/README.md.tmpl", "cmd/server/main.go": "templates/cmd-server-main.go.tmpl", "internal/model/user.go": "templates/internal-model-user.go.tmpl", + "internal/model/errors.go": "templates/internal-model-errors.go.tmpl", + "internal/repository/users.go": "templates/internal-repository-users.go.tmpl", "internal/repository/memory.go": "templates/internal-repository-memory.go.tmpl", "internal/servicecontext/context.go": "templates/internal-servicecontext-context.go.tmpl", "internal/service/user.go": "templates/internal-service-user.go.tmpl", @@ -44,6 +48,13 @@ func Generate(options Options) error { if strings.TrimSpace(options.Module) == "" { return errors.New("module path is required") } + database := strings.ToLower(strings.TrimSpace(options.Database)) + if database == "" { + database = "memory" + } + if database != "memory" && database != "mysql" { + return fmt.Errorf("unsupported database %q: choose memory or mysql", database) + } version := strings.TrimSpace(options.FrameworkVersion) if version == "" { version = currentFrameworkVersion() @@ -72,17 +83,34 @@ func Generate(options Options) error { Module: options.Module, FrameworkVersion: version, FrameworkReplace: replace, + Database: database, } - paths := make([]string, 0, len(files)) - for path := range files { + projectFiles := make(map[string]string, len(files)+3) + for path, source := range files { + projectFiles[path] = source + } + if database == "mysql" { + projectFiles["internal/repository/mysql.go"] = "templates/internal-repository-mysql.go.tmpl" + projectFiles["schema.sql"] = "templates/schema.sql.tmpl" + projectFiles["internal/servicecontext/databaseIntegration_test.go"] = "templates/databaseIntegration_test.go.tmpl" + } + paths := make([]string, 0, len(projectFiles)) + for path := range projectFiles { paths = append(paths, path) } sort.Strings(paths) for _, path := range paths { - content, err := render(files[path], data) + content, err := render(projectFiles[path], data) if err != nil { return fmt.Errorf("render %s: %w", path, err) } + if strings.HasSuffix(path, ".go") { + formatted, err := format.Source([]byte(content)) + if err != nil { + return fmt.Errorf("format %s: %w", path, err) + } + content = string(formatted) + } target := filepath.Join(directory, filepath.FromSlash(path)) if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { return fmt.Errorf("create %s: %w", filepath.Dir(path), err) @@ -105,6 +133,7 @@ type templateData struct { Module string FrameworkVersion string FrameworkReplace string + Database string } func render(name string, data templateData) (string, error) { diff --git a/cmd/go-omni/scaffold/scaffold_test.go b/cmd/go-omni/scaffold/scaffold_test.go index 0c61fb0..d7e1f42 100644 --- a/cmd/go-omni/scaffold/scaffold_test.go +++ b/cmd/go-omni/scaffold/scaffold_test.go @@ -39,7 +39,7 @@ func TestGenerateCreatesServiceLayout(t *testing.T) { if err != nil { t.Fatalf("read generated server: %v", err) } - for _, expected := range []string{"config.LoadFromEnv(\"\")", "logx.SetDefault(logger)", "omni.WithConfig(settings)", "servicecontext.New(logger, users)", "health.New()", "application.RunContext(ctx, settings.Address)"} { + for _, expected := range []string{"config.LoadFromEnv(\"\")", "logx.SetDefault(logger)", "omni.WithConfig(settings)", "servicecontext.New(ctx, logger)", "services.Health.ReadinessHandler()", "application.RunContext(ctx, settings.Address)"} { if !strings.Contains(string(mainContent), expected) { t.Fatalf("generated server missing %q: %s", expected, mainContent) } @@ -56,6 +56,35 @@ func TestGenerateRejectsNonEmptyDirectory(t *testing.T) { } } +func TestGenerateMySQLLayout(t *testing.T) { + project := filepath.Join(t.TempDir(), "billing") + if err := Generate(Options{Dir: project, Module: "example.com/billing", Database: "mysql", FrameworkDir: frameworkRoot(t)}); err != nil { + t.Fatal(err) + } + for _, path := range []string{"schema.sql", "internal/repository/mysql.go", "internal/repository/users.go", "internal/model/errors.go", "internal/servicecontext/databaseIntegration_test.go"} { + if _, err := os.Stat(filepath.Join(project, path)); err != nil { + t.Fatal(err) + } + } + content, err := os.ReadFile(filepath.Join(project, "go.mod")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(content), "github.com/go-sql-driver/mysql v1.9.3") { + t.Fatal("mysql driver version is not pinned") + } +} + +func TestGenerateRejectsUnknownDatabaseBeforeWriting(t *testing.T) { + project := filepath.Join(t.TempDir(), "billing") + if err := Generate(Options{Dir: project, Module: "example.com/billing", Database: "unknown"}); err == nil { + t.Fatal("accepted unknown backend") + } + if _, err := os.Stat(project); !os.IsNotExist(err) { + t.Fatalf("invalid backend created files: %v", err) + } +} + func TestFrameworkReplaceUsesCanonicalPaths(t *testing.T) { root := frameworkRoot(t) project := filepath.Join(t.TempDir(), "generated") diff --git a/cmd/go-omni/scaffold/templates/README.md.tmpl b/cmd/go-omni/scaffold/templates/README.md.tmpl index 83c7e53..a6fd672 100644 --- a/cmd/go-omni/scaffold/templates/README.md.tmpl +++ b/cmd/go-omni/scaffold/templates/README.md.tmpl @@ -4,6 +4,26 @@ This service was generated with `go-omni new`. Framework dependency: `git.yuntuo.hk/k2safe/go-omni {{.FrameworkVersion}}`. +Storage: `{{.Database}}`. + +{{if eq .Database "mysql"}} +先创建业务数据库,再执行项目的建表脚本(替换示例数据库名、账号和地址): + +```sh +mysql -h 127.0.0.1 -u app -p app < schema.sql +``` + +配置数据库连接后启动;初始化失败或缺少表结构时,服务会退出并关闭已创建的连接池: + +```sh +export GO_OMNI_DB_DSN='app:password@tcp(127.0.0.1:3306)/app?parseTime=true' +``` + +应用只检查表结构,不会自动建库、建表或迁移。MySQL 驱动由本项目注册,核心框架仍依赖标准库 `database/sql`。 +{{else}} +默认使用进程内存存储,服务重启后数据清空。需要 MySQL 时,生成新项目时选择 `--database mysql`。 +{{end}} + ```sh go mod tidy go run ./cmd/server @@ -26,10 +46,59 @@ go build -o server ./cmd/server ./server ``` -`GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。健康探针目前没有 -数据库依赖检查;接入数据库后需要向 Checker 注册探针。 +`GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。 +{{if eq .Database "mysql"}} +`/health/ready` 已注册数据库探针:连接失败返回 `503`,数据库恢复后重新返回 `200`。 +`/health/live` 始终只检查服务存活。`GO_OMNI_DB_PING_TIMEOUT` 限制初始化检查和数据库探针的等待时间。 +退出时先排空 HTTP 请求,再关闭数据库连接池,最后关闭日志。 -The generated layout keeps HTTP handlers, business services, repositories, -models and a project-owned ServiceContext separate. Replace the memory -repository with a database implementation when the service is ready for -persistence, then add it to `internal/servicecontext`. +数据库连接池参数: + +```text +GO_OMNI_DB_MAX_OPEN_CONNS=10 +GO_OMNI_DB_MAX_IDLE_CONNS=5 +GO_OMNI_DB_CONN_MAX_LIFETIME=30m +GO_OMNI_DB_CONN_MAX_IDLE_TIME=5m +GO_OMNI_DB_PING_TIMEOUT=5s +``` + +`MAX_IDLE_CONNS=0` 禁用空闲连接,`MAX_OPEN_CONNS=0` 表示不限制连接数。 +项目会启用 MySQL `parseTime` 并按 UTC 读取 DATETIME 字段,JSON 字段保持 lowerCamelCase。 + +事务内复用 Repository(在业务代码中导入 `database/sql` 和项目的 `internal/repository`): + +```go +err := services.DB.InTx(ctx, nil, func(tx *sql.Tx) error { + users := repository.NewMySQLUsers(tx) + _, err := users.Create(ctx, input) + return err +}) +``` + +业务函数返回错误、发生 panic 或 context 取消时会回滚。事务内必须使用基于 `tx` 创建的 Repository。 + +可选集成测试(只针对已应用 schema.sql 的独立测试数据库): + +```sh +GO_OMNI_TEST_MYSQL_DSN='app:password@tcp(127.0.0.1:3306)/test?parseTime=true' go test -race ./... +``` + +未设置 `GO_OMNI_TEST_MYSQL_DSN` 时跳过数据库集成测试。框架仓库中的 `scripts/verifyDatabase.py` +可以自动创建、验收和清理独立的 MySQL 容器。 +{{else}} +内存模式没有外部依赖探针;后续添加依赖时,在 `internal/servicecontext` 注册检查与资源清理。 +{{end}} + +示例接口(无认证,接入实际业务时按项目需求添加认证中间件): + +| 方法 | 路径 | 说明 | +| --- | --- | --- | +| POST | `/users` | 创建,JSON 包含 name、email | +| GET | `/users/:id` | 查询 | +| GET | `/users?page=1&pageSize=20` | 分页 | +| PUT | `/users/:id` | 更新 name、email | +| DELETE | `/users/:id` | 删除,返回 204 | + +接口使用 GoOmni 统一响应,包含 code、message、data 和 requestId。邮箱按小写存储,重复返回 409; +无效输入返回 400,找不到记录返回 404。项目按 Handler / Service / Repository / Model 分层, +基础依赖集中放在 `internal/servicecontext`。 diff --git a/cmd/go-omni/scaffold/templates/cmd-server-main.go.tmpl b/cmd/go-omni/scaffold/templates/cmd-server-main.go.tmpl index 20feffc..e2481f6 100644 --- a/cmd/go-omni/scaffold/templates/cmd-server-main.go.tmpl +++ b/cmd/go-omni/scaffold/templates/cmd-server-main.go.tmpl @@ -10,10 +10,8 @@ import ( omni "git.yuntuo.hk/k2safe/go-omni" "git.yuntuo.hk/k2safe/go-omni/config" - "git.yuntuo.hk/k2safe/go-omni/health" "git.yuntuo.hk/k2safe/go-omni/logx" "{{.Module}}/internal/handler" - "{{.Module}}/internal/repository" "{{.Module}}/internal/service" "{{.Module}}/internal/servicecontext" ) @@ -43,19 +41,17 @@ func run() error { // Access logs and business logs share the same configured sink. logx.SetDefault(logger) - users := repository.NewMemoryUsers() - services := servicecontext.New(logger, users) + services, err := servicecontext.New(ctx, logger) + if err != nil { return err } + defer services.Close() userService := service.NewUsers(services) userHandler := handler.NewUsers(userService) - checker, err := health.New() - if err != nil { - return err - } application := omni.New(omni.WithConfig(settings)) application.OnShutdown(logger.Close) - application.GET("/health/live", checker.LivenessHandler()) - application.GET("/health/ready", checker.ReadinessHandler()) + application.OnShutdown(services.Close) + application.GET("/health/live", services.Health.LivenessHandler()) + application.GET("/health/ready", services.Health.ReadinessHandler()) userHandler.RegisterRoutes(application) return application.RunContext(ctx, settings.Address) } diff --git a/cmd/go-omni/scaffold/templates/databaseIntegration_test.go.tmpl b/cmd/go-omni/scaffold/templates/databaseIntegration_test.go.tmpl new file mode 100644 index 0000000..1c2ff64 --- /dev/null +++ b/cmd/go-omni/scaffold/templates/databaseIntegration_test.go.tmpl @@ -0,0 +1,106 @@ +package servicecontext_test + +import ( + "context" + "database/sql" + "errors" + "fmt" + "io" + "os" + "testing" + "time" + + "git.yuntuo.hk/k2safe/go-omni/database" + "git.yuntuo.hk/k2safe/go-omni/logx" + "{{.Module}}/internal/model" + "{{.Module}}/internal/repository" + "{{.Module}}/internal/servicecontext" +) + +// These tests require a dedicated database with schema.sql applied. Ordinary +// go test runs skip them; scripts/verifyDatabase.py provisions an isolated DB. +func databaseContext(t *testing.T) *servicecontext.Context { + t.Helper() + dsn := os.Getenv("GO_OMNI_TEST_MYSQL_DSN") + if dsn == "" { t.Skip("set GO_OMNI_TEST_MYSQL_DSN to a dedicated test database") } + for _, key := range []string{"MAX_OPEN_CONNS", "MAX_IDLE_CONNS", "CONN_MAX_LIFETIME", "CONN_MAX_IDLE_TIME", "PING_TIMEOUT"} { + t.Setenv("GO_OMNI_DB_"+key, "") + } + t.Setenv("GO_OMNI_DB_DSN", dsn) + t.Setenv("GO_OMNI_DB_DRIVER", "mysql") + logger, err := logx.New(logx.Config{Writer: io.Discard}) + if err != nil { t.Fatal(err) } + t.Cleanup(func() { _ = logger.Close() }) + services, err := servicecontext.New(context.Background(), logger) + if err != nil { t.Fatal(err) } + t.Cleanup(func() { _ = services.Close() }) + return services +} + +func TestDatabaseTransactions(t *testing.T) { + services := databaseContext(t) + ctx := context.Background() + for _, mode := range []string{"commit", "error", "panic", "cancel"} { + t.Run(mode, func(t *testing.T) { + email := fmt.Sprintf("transaction-%d-%s@example.com", time.Now().UnixNano(), mode) + t.Cleanup(func() { _, _ = services.DB.ExecContext(ctx, "DELETE FROM users WHERE email = ?", email) }) + txCtx, cancel := context.WithCancel(ctx) + defer cancel() + abort := errors.New("abort business operation") + var err error + var recovered any + func() { + defer func() { recovered = recover() }() + err = services.DB.InTx(txCtx, nil, func(tx *sql.Tx) error { + users := repository.NewMySQLUsers(tx) + if _, err := users.Create(txCtx, model.UserInput{Name: "transaction", Email: email}); err != nil { return err } + switch mode { + case "error": return abort + case "panic": panic(abort) + case "cancel": cancel(); return txCtx.Err() + } + return nil + }) + }() + switch mode { + case "commit": if err != nil || recovered != nil { t.Fatalf("commit: error=%v panic=%v", err, recovered) } + case "error": if !errors.Is(err, abort) { t.Fatalf("lost business error: %v", err) } + case "panic": if recovered != abort { t.Fatalf("lost panic: %v", recovered) } + case "cancel": if !errors.Is(err, context.Canceled) { t.Fatalf("lost cancellation: %v", err) } + } + var count int + if err := services.DB.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE email = ?", email).Scan(&count); err != nil { t.Fatal(err) } + want := 0 + if mode == "commit" { want = 1 } + if count != want { t.Fatalf("%s persisted %d rows, want %d", mode, count, want) } + }) + } +} + +func TestDatabasePoolHealthAndClose(t *testing.T) { + services := databaseContext(t) + ctx := context.Background() + if result := services.Health.Check(ctx); !result.Ready() || len(result.Checks) != 1 { t.Fatalf("missing database readiness: %+v", result) } + + cfg := services.DB.Config() + cfg.MaxOpenConns, cfg.MaxIdleConns = 1, 0 + pool, err := database.Open(ctx, cfg) + if err != nil { t.Fatal(err) } + if idle := pool.Stats().Idle; idle != 0 { t.Fatalf("zero idle config retained %d connections", idle) } + _ = pool.Close() + + cfg.MaxIdleConns, cfg.PingTimeout = 1, 50*time.Millisecond + pool, err = database.Open(ctx, cfg) + if err != nil { t.Fatal(err) } + defer pool.Close() + connection, err := pool.Conn(ctx) + if err != nil { t.Fatal(err) } + defer connection.Close() + started := time.Now() + err = pool.Health(ctx) + if !errors.Is(err, context.DeadlineExceeded) || time.Since(started) > time.Second { t.Fatalf("health did not bound pool waiting: %v", err) } + + if err := services.Close(); err != nil { t.Fatal(err) } + if err := services.Close(); err != nil { t.Fatal(err) } + if result := services.Health.Check(ctx); result.Ready() { t.Fatal("closed database was still ready") } +} diff --git a/cmd/go-omni/scaffold/templates/go.mod.tmpl b/cmd/go-omni/scaffold/templates/go.mod.tmpl index 603118b..2462e54 100644 --- a/cmd/go-omni/scaffold/templates/go.mod.tmpl +++ b/cmd/go-omni/scaffold/templates/go.mod.tmpl @@ -3,6 +3,8 @@ module {{.Module}} go 1.22 require git.yuntuo.hk/k2safe/go-omni {{.FrameworkVersion}} +{{if eq .Database "mysql"}}require github.com/go-sql-driver/mysql v1.9.3 +{{end}} {{- if .FrameworkReplace}} replace git.yuntuo.hk/k2safe/go-omni => {{.FrameworkReplace}} {{- end}} diff --git a/cmd/go-omni/scaffold/templates/internal-handler-user.go.tmpl b/cmd/go-omni/scaffold/templates/internal-handler-user.go.tmpl index 5c2477e..e207b20 100644 --- a/cmd/go-omni/scaffold/templates/internal-handler-user.go.tmpl +++ b/cmd/go-omni/scaffold/templates/internal-handler-user.go.tmpl @@ -1,9 +1,10 @@ package handler import ( + "errors" "net/http" - omni "git.yuntuo.hk/k2safe/go-omni" + "git.yuntuo.hk/k2safe/go-omni" "{{.Module}}/internal/model" "{{.Module}}/internal/service" ) @@ -12,20 +13,109 @@ type Users struct { service *service.Users } -func NewUsers(service *service.Users) *Users { return &Users{service: service} } +func NewUsers(service *service.Users) *Users { + return &Users{service: service} +} func (h *Users) RegisterRoutes(application *omni.App) { - application.POST("/users", h.Create) + users := application.Group("/users") + users.POST("", h.Create) + users.GET("", h.List) + users.GET("/:id", h.Get) + users.PUT("/:id", h.Update) + users.DELETE("/:id", h.Delete) +} + +// HTTP-specific validation tags stay here, outside the business model. +type userBody struct { + Name string `json:"name" binding:"required,max=100"` + Email string `json:"email" binding:"required,email,max=254"` +} + +type userPath struct { + ID int64 `uri:"id" binding:"gt=0"` +} + +type listQuery struct { + Page int `form:"page,default=1" binding:"gte=1"` + PageSize int `form:"pageSize,default=20" binding:"gte=1,lte=100"` } func (h *Users) Create(ctx *omni.Context) error { - var input model.UserInput - if err := ctx.BindJSON(&input); err != nil { + var body userBody + if err := ctx.BindJSON(&body); err != nil { return err } - user, err := h.service.Create(ctx.Context(), input) + user, err := h.service.Create(ctx.Request().Context(), model.UserInput(body)) if err != nil { + return userError(err) + } + return ctx.Created(user) +} + +func (h *Users) Get(ctx *omni.Context) error { + var path userPath + if err := ctx.BindURI(&path); err != nil { + return err + } + user, err := h.service.Get(ctx.Request().Context(), path.ID) + if err != nil { + return userError(err) + } + return ctx.OK(user) +} + +func (h *Users) List(ctx *omni.Context) error { + var query listQuery + if err := ctx.BindQuery(&query); err != nil { + return err + } + page, err := h.service.List(ctx.Request().Context(), query.Page, query.PageSize) + if err != nil { + return userError(err) + } + return ctx.OK(page) +} + +func (h *Users) Update(ctx *omni.Context) error { + var path userPath + if err := ctx.BindURI(&path); err != nil { + return err + } + var body userBody + if err := ctx.BindJSON(&body); err != nil { + return err + } + user, err := h.service.Update(ctx.Request().Context(), path.ID, model.UserInput(body)) + if err != nil { + return userError(err) + } + return ctx.OK(user) +} + +func (h *Users) Delete(ctx *omni.Context) error { + var path userPath + if err := ctx.BindURI(&path); err != nil { + return err + } + if err := h.service.Delete(ctx.Request().Context(), path.ID); err != nil { + return userError(err) + } + return ctx.NoContent() +} + +func userError(err error) error { + switch { + case errors.Is(err, model.ErrUserNotFound): + return omni.NotFound("USER_NOT_FOUND", "user was not found").WithCause(err) + case errors.Is(err, model.ErrEmailTaken): + return omni.NewError(http.StatusConflict, "EMAIL_TAKEN", "email is already in use").WithCause(err) + case errors.Is(err, model.ErrInvalidUser): + return omni.BadRequest("INVALID_USER", "name or email is invalid").WithCause(err) + case errors.Is(err, model.ErrInvalidPage): + return omni.BadRequest("INVALID_QUERY", "pagination is invalid").WithCause(err) + default: + // GoOmni masks unexpected errors; internal details stay in server logs. return err } - return ctx.JSON(http.StatusCreated, user) } diff --git a/cmd/go-omni/scaffold/templates/internal-model-errors.go.tmpl b/cmd/go-omni/scaffold/templates/internal-model-errors.go.tmpl new file mode 100644 index 0000000..3f6e7c8 --- /dev/null +++ b/cmd/go-omni/scaffold/templates/internal-model-errors.go.tmpl @@ -0,0 +1,11 @@ +package model + +import "errors" + +// Domain errors are translated into HTTP responses only by the handler layer. +var ( + ErrUserNotFound = errors.New("user not found") + ErrEmailTaken = errors.New("email already in use") + ErrInvalidUser = errors.New("invalid user") + ErrInvalidPage = errors.New("invalid pagination") +) diff --git a/cmd/go-omni/scaffold/templates/internal-model-user.go.tmpl b/cmd/go-omni/scaffold/templates/internal-model-user.go.tmpl index f810df0..376f3eb 100644 --- a/cmd/go-omni/scaffold/templates/internal-model-user.go.tmpl +++ b/cmd/go-omni/scaffold/templates/internal-model-user.go.tmpl @@ -2,6 +2,7 @@ package model import "time" +// User is the business entity; it has no dependency on HTTP or a database driver. type User struct { ID int64 `json:"id"` Name string `json:"name"` @@ -10,7 +11,15 @@ type User struct { UpdatedAt time.Time `json:"updatedAt"` } +// UserInput is shared by creation and full replacement (PUT). type UserInput struct { - Name string `json:"name"` - Email string `json:"email"` + Name string + Email string +} + +type UserPage struct { + Items []User `json:"items"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"pageSize"` } diff --git a/cmd/go-omni/scaffold/templates/internal-repository-memory.go.tmpl b/cmd/go-omni/scaffold/templates/internal-repository-memory.go.tmpl index 06d8f1f..f1d319f 100644 --- a/cmd/go-omni/scaffold/templates/internal-repository-memory.go.tmpl +++ b/cmd/go-omni/scaffold/templates/internal-repository-memory.go.tmpl @@ -2,30 +2,125 @@ package repository import ( "context" + "sort" "sync" "time" "{{.Module}}/internal/model" ) -type UserRepository interface { - Create(context.Context, model.UserInput) (model.User, error) -} - +// MemoryUsers stores data only for this process. Replace it with a database +// implementation of UserRepository when persistence is required. type MemoryUsers struct { - mu sync.Mutex + mu sync.RWMutex + users map[int64]model.User + emails map[string]int64 nextID int64 - users []model.User } -func NewMemoryUsers() *MemoryUsers { return &MemoryUsers{nextID: 1} } +func NewMemoryUsers() *MemoryUsers { + return &MemoryUsers{ + users: make(map[int64]model.User), + emails: make(map[string]int64), + nextID: 1, + } +} -func (r *MemoryUsers) Create(_ context.Context, input model.UserInput) (model.User, error) { +func (r *MemoryUsers) Create(ctx context.Context, input model.UserInput) (model.User, error) { r.mu.Lock() defer r.mu.Unlock() + if err := ctx.Err(); err != nil { + return model.User{}, err + } + if _, exists := r.emails[input.Email]; exists { + return model.User{}, model.ErrEmailTaken + } now := time.Now().UTC() - user := model.User{ID: r.nextID, Name: input.Name, Email: input.Email, CreatedAt: now, UpdatedAt: now} + user := model.User{ + ID: r.nextID, Name: input.Name, Email: input.Email, + CreatedAt: now, UpdatedAt: now, + } r.nextID++ - r.users = append(r.users, user) + r.users[user.ID] = user + r.emails[user.Email] = user.ID return user, nil } + +func (r *MemoryUsers) Get(ctx context.Context, id int64) (model.User, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if err := ctx.Err(); err != nil { + return model.User{}, err + } + user, exists := r.users[id] + if !exists { + return model.User{}, model.ErrUserNotFound + } + return user, nil +} + +func (r *MemoryUsers) List(ctx context.Context, page, pageSize int) (model.UserPage, error) { + if page < 1 || pageSize < 1 || pageSize > 100 { + return model.UserPage{}, model.ErrInvalidPage + } + r.mu.RLock() + if err := ctx.Err(); err != nil { + r.mu.RUnlock() + return model.UserPage{}, err + } + users := make([]model.User, 0, len(r.users)) + for _, user := range r.users { + users = append(users, user) + } + r.mu.RUnlock() + + // A stable order is essential: Go map iteration order changes between calls. + sort.Slice(users, func(i, j int) bool { return users[i].ID < users[j].ID }) + result := model.UserPage{ + Items: []model.User{}, Total: len(users), Page: page, PageSize: pageSize, + } + // Check the page before multiplying so even a very large page cannot overflow. + if len(users) == 0 || page-1 > (len(users)-1)/pageSize { + return result, nil + } + start := (page - 1) * pageSize + count := min(pageSize, len(users)-start) + result.Items = users[start : start+count] + return result, nil +} + +func (r *MemoryUsers) Update(ctx context.Context, id int64, input model.UserInput) (model.User, error) { + r.mu.Lock() + defer r.mu.Unlock() + if err := ctx.Err(); err != nil { + return model.User{}, err + } + user, exists := r.users[id] + if !exists { + return model.User{}, model.ErrUserNotFound + } + if owner, exists := r.emails[input.Email]; exists && owner != id { + return model.User{}, model.ErrEmailTaken + } + delete(r.emails, user.Email) + user.Name, user.Email = input.Name, input.Email + user.UpdatedAt = time.Now().UTC() + r.users[id] = user + r.emails[user.Email] = id + return user, nil +} + +func (r *MemoryUsers) Delete(ctx context.Context, id int64) error { + r.mu.Lock() + defer r.mu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + user, exists := r.users[id] + if !exists { + return model.ErrUserNotFound + } + delete(r.users, id) + delete(r.emails, user.Email) + return nil +} diff --git a/cmd/go-omni/scaffold/templates/internal-repository-mysql.go.tmpl b/cmd/go-omni/scaffold/templates/internal-repository-mysql.go.tmpl new file mode 100644 index 0000000..f78a5e6 --- /dev/null +++ b/cmd/go-omni/scaffold/templates/internal-repository-mysql.go.tmpl @@ -0,0 +1,139 @@ +package repository + +import ( + "context" + "database/sql" + "errors" + + "{{.Module}}/internal/model" + "github.com/go-sql-driver/mysql" +) + +// SQLExecutor is implemented by both *sql.DB and *sql.Tx. A repository created +// with a transaction keeps all of its queries on that same transaction. +type SQLExecutor interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) + QueryContext(context.Context, string, ...any) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +// MySQLUsers is the persistent implementation of UserRepository. +// The schema is documented in schema.sql; migrations stay outside the framework. +type MySQLUsers struct { + db SQLExecutor +} + +func NewMySQLUsers(db SQLExecutor) *MySQLUsers { + return &MySQLUsers{db: db} +} + +// CheckSchema validates that the application's migration has been applied. +// Runtime startup never creates or alters tables. +func (r *MySQLUsers) CheckSchema(ctx context.Context) error { + rows, err := r.db.QueryContext(ctx, `SELECT id, name, email, created_at, updated_at FROM users LIMIT 0`) + if err != nil { return err } + return rows.Close() +} + +func (r *MySQLUsers) Create(ctx context.Context, input model.UserInput) (model.User, error) { + result, err := r.db.ExecContext(ctx, ` + INSERT INTO users (name, email, created_at, updated_at) + VALUES (?, ?, UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))`, input.Name, input.Email) + if err != nil { + return model.User{}, mapMySQLError(err) + } + id, err := result.LastInsertId() + if err != nil { + return model.User{}, err + } + return r.Get(ctx, id) +} + +func (r *MySQLUsers) Get(ctx context.Context, id int64) (model.User, error) { + var user model.User + err := r.db.QueryRowContext(ctx, ` + SELECT id, name, email, created_at, updated_at + FROM users WHERE id = ?`, id).Scan( + &user.ID, &user.Name, &user.Email, &user.CreatedAt, &user.UpdatedAt, + ) + if errors.Is(err, sql.ErrNoRows) { + return model.User{}, model.ErrUserNotFound + } + if err != nil { + return model.User{}, err + } + return user, nil +} + +func (r *MySQLUsers) List(ctx context.Context, page, pageSize int) (model.UserPage, error) { + if page < 1 || pageSize < 1 || pageSize > 100 { return model.UserPage{}, model.ErrInvalidPage } + var total int + if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM users`).Scan(&total); err != nil { + return model.UserPage{}, err + } + result := model.UserPage{Items: []model.User{}, Total: total, Page: page, PageSize: pageSize} + if total == 0 || page-1 > (total-1)/pageSize { + return result, nil + } + offset := (page - 1) * pageSize + rows, err := r.db.QueryContext(ctx, ` + SELECT id, name, email, created_at, updated_at + FROM users ORDER BY id LIMIT ? OFFSET ?`, pageSize, offset) + if err != nil { + return model.UserPage{}, err + } + defer rows.Close() + for rows.Next() { + var user model.User + if err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.CreatedAt, &user.UpdatedAt); err != nil { + return model.UserPage{}, err + } + result.Items = append(result.Items, user) + } + if err := rows.Err(); err != nil { + return model.UserPage{}, err + } + return result, nil +} + +func (r *MySQLUsers) Update(ctx context.Context, id int64, input model.UserInput) (model.User, error) { + result, err := r.db.ExecContext(ctx, ` + UPDATE users SET name = ?, email = ?, updated_at = UTC_TIMESTAMP(6) + WHERE id = ?`, input.Name, input.Email, id) + if err != nil { + return model.User{}, mapMySQLError(err) + } + changed, err := result.RowsAffected() + if err != nil { + return model.User{}, err + } + if changed == 0 { + // MySQL may report zero affected rows when the replacement is identical; + // distinguish that from an ID that does not exist. + return r.Get(ctx, id) + } + return r.Get(ctx, id) +} + +func (r *MySQLUsers) Delete(ctx context.Context, id int64) error { + result, err := r.db.ExecContext(ctx, `DELETE FROM users WHERE id = ?`, id) + if err != nil { + return err + } + deleted, err := result.RowsAffected() + if err != nil { + return err + } + if deleted == 0 { + return model.ErrUserNotFound + } + return nil +} + +func mapMySQLError(err error) error { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return model.ErrEmailTaken + } + return err +} diff --git a/cmd/go-omni/scaffold/templates/internal-repository-users.go.tmpl b/cmd/go-omni/scaffold/templates/internal-repository-users.go.tmpl new file mode 100644 index 0000000..517d6b9 --- /dev/null +++ b/cmd/go-omni/scaffold/templates/internal-repository-users.go.tmpl @@ -0,0 +1,16 @@ +package repository + +import ( +"context" +"{{.Module}}/internal/model" +) + +// UserRepository describes the storage operations the business needs. +// Implementations must atomically enforce unique, normalized email addresses. +type UserRepository interface { + Create(context.Context, model.UserInput) (model.User, error) + Get(context.Context, int64) (model.User, error) + List(context.Context, int, int) (model.UserPage, error) + Update(context.Context, int64, model.UserInput) (model.User, error) + Delete(context.Context, int64) error +} diff --git a/cmd/go-omni/scaffold/templates/internal-service-user.go.tmpl b/cmd/go-omni/scaffold/templates/internal-service-user.go.tmpl index 6cdcb52..ea124cb 100644 --- a/cmd/go-omni/scaffold/templates/internal-service-user.go.tmpl +++ b/cmd/go-omni/scaffold/templates/internal-service-user.go.tmpl @@ -2,20 +2,61 @@ package service import ( "context" + "net/mail" + "strings" + "unicode/utf8" "{{.Module}}/internal/model" "{{.Module}}/internal/servicecontext" ) type Users struct { - context *servicecontext.Context + services *servicecontext.Context } -func NewUsers(context *servicecontext.Context) *Users { - return &Users{context: context} +func NewUsers(services *servicecontext.Context) *Users { + return &Users{services: services} } func (s *Users) Create(ctx context.Context, input model.UserInput) (model.User, error) { - // Put business normalization and rules here. - return s.context.Users.Create(ctx, input) + input, err := normalize(input) + if err != nil { + return model.User{}, err + } + return s.services.Users.Create(ctx, input) +} + +func (s *Users) Get(ctx context.Context, id int64) (model.User, error) { + return s.services.Users.Get(ctx, id) +} + +func (s *Users) List(ctx context.Context, page, pageSize int) (model.UserPage, error) { + if page < 1 || pageSize < 1 || pageSize > 100 { + return model.UserPage{}, model.ErrInvalidPage + } + return s.services.Users.List(ctx, page, pageSize) +} + +func (s *Users) Update(ctx context.Context, id int64, input model.UserInput) (model.User, error) { + input, err := normalize(input) + if err != nil { + return model.User{}, err + } + return s.services.Users.Update(ctx, id, input) +} + +func (s *Users) Delete(ctx context.Context, id int64) error { + return s.services.Users.Delete(ctx, id) +} + +func normalize(input model.UserInput) (model.UserInput, error) { + input.Name = strings.TrimSpace(input.Name) + // This example treats the entire email address as case-insensitive. + input.Email = strings.ToLower(strings.TrimSpace(input.Email)) + address, err := mail.ParseAddress(input.Email) + if input.Name == "" || utf8.RuneCountInString(input.Name) > 100 || + len(input.Email) > 254 || err != nil || address.Address != input.Email { + return model.UserInput{}, model.ErrInvalidUser + } + return input, nil } diff --git a/cmd/go-omni/scaffold/templates/internal-servicecontext-context.go.tmpl b/cmd/go-omni/scaffold/templates/internal-servicecontext-context.go.tmpl index a64b1b0..ade1abc 100644 --- a/cmd/go-omni/scaffold/templates/internal-servicecontext-context.go.tmpl +++ b/cmd/go-omni/scaffold/templates/internal-servicecontext-context.go.tmpl @@ -1,6 +1,12 @@ package servicecontext import ( + "context" + {{if eq .Database "mysql"}}"fmt" + "time" + "git.yuntuo.hk/k2safe/go-omni/database" + "github.com/go-sql-driver/mysql"{{end}} + "git.yuntuo.hk/k2safe/go-omni/health" "git.yuntuo.hk/k2safe/go-omni/logx" "{{.Module}}/internal/repository" ) @@ -10,8 +16,51 @@ import ( type Context struct { Logger *logx.Logger Users repository.UserRepository + Health *health.Checker + {{if eq .Database "mysql"}}DB *database.DB{{end}} } -func New(logger *logx.Logger, users repository.UserRepository) *Context { - return &Context{Logger: logger, Users: users} +// New initializes the project's dependencies before the HTTP listener opens. +// The caller owns Logger; Close releases only resources created here. +func New(ctx context.Context, logger *logx.Logger) (*Context, error) { + if ctx == nil { ctx = context.Background() } + if err := ctx.Err(); err != nil { return nil, err } + checker, err := health.New() + if err != nil { return nil, err } + services := &Context{Logger: logger, Health: checker} + {{if eq .Database "mysql"}} + cfg, err := database.ConfigFromEnv("") + if err != nil { return nil, err } + if cfg.Driver != "mysql" { return nil, fmt.Errorf("this project requires GO_OMNI_DB_DRIVER=mysql") } + dsn, err := mysql.ParseDSN(cfg.DSN) + if err != nil { return nil, fmt.Errorf("parse mysql connection settings: %w", err) } + if dsn.DBName == "" { return nil, fmt.Errorf("mysql connection settings require a database name") } + // DATETIME columns are returned as time.Time in UTC in the business model. + dsn.ParseTime = true + dsn.Loc = time.UTC + cfg.DSN = dsn.FormatDSN() + db, err := database.Open(ctx, cfg) + if err != nil { return nil, err } + users := repository.NewMySQLUsers(db.DB) + checkCtx, cancel := context.WithTimeout(ctx, cfg.PingTimeout) + defer cancel() + if err := users.CheckSchema(checkCtx); err != nil { + _ = db.Close() + return nil, fmt.Errorf("check users schema (apply schema.sql before starting): %w", err) + } + if err := checker.Add(health.Probe{Name: "database", Timeout: cfg.PingTimeout, Check: db.Health}); err != nil { + _ = db.Close() + return nil, err + } + services.DB, services.Users = db, users + {{else}} + services.Users = repository.NewMemoryUsers() + {{end}} + return services, nil +} + +// Close is safe to call on startup failure and again during application exit. +func (c *Context) Close() error { + {{if eq .Database "mysql"}}if c != nil && c.DB != nil { return c.DB.Close() }{{end}} + return nil } diff --git a/cmd/go-omni/scaffold/templates/schema.sql.tmpl b/cmd/go-omni/scaffold/templates/schema.sql.tmpl new file mode 100644 index 0000000..0219b5d --- /dev/null +++ b/cmd/go-omni/scaffold/templates/schema.sql.tmpl @@ -0,0 +1,9 @@ +CREATE TABLE users ( + id BIGINT NOT NULL AUTO_INCREMENT, + name VARCHAR(100) NOT NULL, + email VARCHAR(254) NOT NULL, + created_at DATETIME(6) NOT NULL, + updated_at DATETIME(6) NOT NULL, + PRIMARY KEY (id), + UNIQUE KEY uq_users_email (email) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/cmd/go-omni/scaffold/version.go b/cmd/go-omni/scaffold/version.go index 1c64700..e700bb8 100644 --- a/cmd/go-omni/scaffold/version.go +++ b/cmd/go-omni/scaffold/version.go @@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni" // Source builds can carry an unpublished VCS version (including +dirty). // Keep this fallback in sync with the next release; installed CLIs use their // downloaded module version, which also has a module checksum. -const defaultFrameworkVersion = "v0.1.1" +const defaultFrameworkVersion = "v0.2.0" var frameworkVersionPattern = regexp.MustCompile(`^v[01]\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)(-((0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)(\.(0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*))?(\+[0-9A-Za-z-]+(\.[0-9A-Za-z-]+)*)?$`) diff --git a/database/README.md b/database/README.md index 985c026..7f53821 100644 --- a/database/README.md +++ b/database/README.md @@ -4,7 +4,7 @@ - `Config` / `ConfigFromEnv`:连接池参数和 `GO_OMNI_DB_*` 环境变量。 - `Open`:创建连接池、设置连接上限并在返回前执行 `PingContext`。 -- `Health`:用于就绪检查。 +- `Health`:用于就绪检查,等待连接和 Ping 均受 `PingTimeout` 限制。 - `InTx`:统一事务提交和回滚;工作函数返回错误或 panic 时回滚。 - `Close`:由嵌入的 `*sql.DB` 提供,应用退出时显式调用。 @@ -18,7 +18,13 @@ import ( ) cfg, err := database.ConfigFromEnv("GO_OMNI_DB_") +if err != nil { + return err +} db, err := database.Open(context.Background(), cfg) +if err != nil { + return err +} defer db.Close() ``` @@ -35,3 +41,18 @@ GO_OMNI_DB_PING_TIMEOUT=5s ``` `DSN` 必填,驱动默认是 `mysql`。数据库建库、迁移、表结构和具体 Repository 属于业务项目或可选模块,不会被核心包自动执行。 + +`MaxIdleConns=0` 会禁用空闲连接。初始化 Ping 失败会关闭已创建的连接池。 + +`go-omni new --database mysql` 会生成项目自己的 MySQL Repository、ServiceContext、 +数据库就绪探针和 `schema.sql`。核心包不需要知道业务表结构。 + +真实数据库验收(需要本地 Docker): + +```sh +python3 scripts/verifyDatabase.py --version v0.2.0 +``` + +脚本仅操作本次创建的独立容器与数据库,验证 HTTP CRUD、服务重启持久化、事务提交与 +错误/panic/取消回滚、连接池关闭、缺少表结构时启动失败,以及停库后的 readiness 503 和恢复后的 200。 +结束时清理容器、匿名数据卷和临时项目。常规 `go test ./...` 不依赖 Docker。 diff --git a/database/database.go b/database/database.go index 95c5928..9986239 100644 --- a/database/database.go +++ b/database/database.go @@ -93,6 +93,9 @@ type DB struct { } func Open(ctx context.Context, cfg Config) (*DB, error) { + if ctx == nil { + ctx = context.Background() + } if err := cfg.Validate(); err != nil { return nil, err } @@ -103,9 +106,9 @@ func Open(ctx context.Context, cfg Config) (*DB, error) { if cfg.MaxOpenConns > 0 { db.SetMaxOpenConns(cfg.MaxOpenConns) } - if cfg.MaxIdleConns > 0 { - db.SetMaxIdleConns(cfg.MaxIdleConns) - } + // database/sql uses zero to disable idle connections; do not leave its + // default pool size in place when the caller explicitly chooses zero. + db.SetMaxIdleConns(cfg.MaxIdleConns) db.SetConnMaxLifetime(cfg.ConnMaxLifetime) db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime) checkCtx, cancel := context.WithTimeout(ctx, cfg.PingTimeout) @@ -123,7 +126,12 @@ func (db *DB) Health(ctx context.Context) error { if db == nil || db.DB == nil { return errors.New("database is not initialized") } - return db.PingContext(ctx) + if ctx == nil { + ctx = context.Background() + } + checkCtx, cancel := context.WithTimeout(ctx, db.config.PingTimeout) + defer cancel() + return db.PingContext(checkCtx) } // InTx commits only when work returns nil. Any work error or panic rolls back. diff --git a/database/lifecycle_test.go b/database/lifecycle_test.go new file mode 100644 index 0000000..7913a24 --- /dev/null +++ b/database/lifecycle_test.go @@ -0,0 +1,39 @@ +package database + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "sync/atomic" + "testing" +) + +var failedPingClosed atomic.Int32 +var errPingUnavailable = errors.New("database unavailable") + +type failedPingDriver struct{} +type failedPingConnection struct{} + +func (failedPingDriver) Open(string) (driver.Conn, error) { return failedPingConnection{}, nil } +func (failedPingConnection) Prepare(string) (driver.Stmt, error) { + return nil, errors.New("not implemented") +} +func (failedPingConnection) Begin() (driver.Tx, error) { return nil, errors.New("not implemented") } +func (failedPingConnection) Close() error { failedPingClosed.Add(1); return nil } +func (failedPingConnection) Ping(context.Context) error { return errPingUnavailable } + +func init() { sql.Register("goOmniFailPing", failedPingDriver{}) } + +func TestOpenClosesPoolWhenStartupPingFails(t *testing.T) { + failedPingClosed.Store(0) + cfg := DefaultConfig() + cfg.Driver, cfg.DSN = "goOmniFailPing", "test" + pool, err := Open(context.Background(), cfg) + if pool != nil || !errors.Is(err, errPingUnavailable) { + t.Fatalf("unexpected startup result: db=%v err=%v", pool, err) + } + if closed := failedPingClosed.Load(); closed != 1 { + t.Fatalf("startup failure closed %d connections, want 1", closed) + } +} diff --git a/scripts/verifyDatabase.py b/scripts/verifyDatabase.py new file mode 100644 index 0000000..4c0ef3f --- /dev/null +++ b/scripts/verifyDatabase.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 +"""Verify a generated MySQL service against a disposable local Docker database.""" + +import argparse +import json +import os +from pathlib import Path +import secrets +import signal +import socket +import subprocess +import tempfile +import time +import urllib.error +import urllib.request +import uuid + +from verifyScaffold import MODULE, run + + +def waitFor(check, description, timeout=20): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if check(): + return + time.sleep(0.1) + raise AssertionError(f"timed out: {description}") + + +class MySQL: + def __init__(self, directory, env, image): + self.directory, self.env, self.image = directory, dict(env), image + self.name = "go-omni-check-" + uuid.uuid4().hex[:12] + self.container = None + self.password = secrets.token_hex(18) + self.env.update(MYSQL_ROOT_PASSWORD=secrets.token_hex(18), MYSQL_PASSWORD=self.password) + self.env["MYSQL_PWD"] = self.env["MYSQL_ROOT_PASSWORD"] + + def sql(self, statement, database=None): + args = ["docker", "exec", "-i", "--env", "MYSQL_PWD", self.container, + "mysql", "--protocol=tcp", "-h127.0.0.1", "-uroot", "--batch", "--skip-column-names"] + if database: + args.append(database) + result = subprocess.run(args, cwd=self.directory, env=self.env, input=statement, + text=True, capture_output=True, timeout=10) + if result.returncode: + raise RuntimeError(f"test database SQL failed: {result.stderr}") + return result.stdout.strip() + + def wait(self): + def ready(): + try: + return self.sql("SELECT 1") == "1" + except (RuntimeError, subprocess.TimeoutExpired): + return False + waitFor(ready, "MySQL ready", timeout=90) + + def start(self): + endpoint = self.env.get("DOCKER_HOST") or run( + ["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"], + self.directory, self.env).strip() + if not endpoint.startswith(("unix://", "npipe://")): + raise RuntimeError("database verification requires a local Docker context") + # Docker may assign a different random published port after restart. + # Choose the loopback port once so outage/recovery keeps the same DSN. + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + hostPort = listener.getsockname()[1] + print("Starting isolated MySQL container", flush=True) + self.container = run(["docker", "run", "--detach", "--name", self.name, + "--label", "go-omni.test=database", "--publish", f"127.0.0.1:{hostPort}:3306", + "--env", "MYSQL_ROOT_PASSWORD", "--env", "MYSQL_PASSWORD", + "--env", "MYSQL_DATABASE=omni_test", "--env", "MYSQL_USER=omni_test", + self.image], self.directory, self.env).strip() + self.wait() + port = run(["docker", "port", self.container, "3306/tcp"], self.directory, self.env).strip().rsplit(":", 1)[1] + self.dsn = (f"omni_test:{self.password}@tcp(127.0.0.1:{port})/omni_test" + "?parseTime=true&timeout=2s&readTimeout=2s&writeTimeout=2s") + self.sql("CREATE DATABASE omni_empty; GRANT ALL ON omni_empty.* TO 'omni_test'@'%';") + + def close(self): + if self.container: + # Remove only the container and anonymous volume created by this run. + run(["docker", "rm", "--force", "--volumes", self.container], self.directory, self.env) + + +class Server: + def __init__(self, binary, directory, env): + self.binary, self.directory, self.env = binary, directory, dict(env) + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + self.address = f"127.0.0.1:{listener.getsockname()[1]}" + self.env["GO_OMNI_HTTP_ADDR"] = self.address + self.process = None + self.logPath = directory / f"server-{uuid.uuid4().hex[:8]}.log" + self.client = urllib.request.build_opener(urllib.request.ProxyHandler({})) + + def request(self, method, path, body=None, expected=200): + req = urllib.request.Request(f"http://{self.address}{path}", method=method, + data=None if body is None else json.dumps(body).encode(), + headers={"Content-Type": "application/json", "X-Request-ID": "databaseCheck"}) + try: + response = self.client.open(req, timeout=3) + except urllib.error.HTTPError as error: + response = error + with response: + raw = response.read() + if response.status != expected: + raise AssertionError(f"{method} {path}: expected {expected}, got {response.status}: {raw!r}") + if response.headers.get("X-Request-ID") != "databaseCheck": + raise AssertionError("missing requestId header") + if not raw: + return None + result = json.loads(raw) + if result.get("requestId") != "databaseCheck": + raise AssertionError("missing camelCase requestId") + return result + + def start(self): + with self.logPath.open("w") as logs: + self.process = subprocess.Popen([str(self.binary)], cwd=self.directory, env=self.env, + stdout=logs, stderr=subprocess.STDOUT) + def ready(): + if self.process.poll() is not None: + raise RuntimeError(f"service exited: {self.logPath.read_text()}") + try: + return self.request("GET", "/health/live")["data"]["status"] == "ok" + except (OSError, urllib.error.URLError): + return False + waitFor(ready, "service startup") + + def stop(self): + if self.process and self.process.poll() is None: + self.process.send_signal(signal.SIGTERM) + code = self.process.wait(timeout=10) + if code != 0: + raise AssertionError(f"service exit {code}: {self.logPath.read_text()}") + + def close(self): + if self.process and self.process.poll() is None: + self.process.kill() + self.process.wait(timeout=5) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--go", default="go") + parser.add_argument("--version", default="v0.2.0") + parser.add_argument("--cli", help="Use a locally built CLI") + parser.add_argument("--framework-dir", help="Local framework source for pre-release checks") + parser.add_argument("--image", default="mysql:8.4") + args = parser.parse_args() + env = {key: value for key, value in os.environ.items() if not key.startswith("GO_OMNI_")} + env["GOWORK"] = "off" + env["GOPRIVATE"] = ",".join(filter(None, [env.get("GOPRIVATE"), "git.yuntuo.hk"])) + with tempfile.TemporaryDirectory(prefix="go omni database ") as temporary: + directory = Path(temporary).resolve() + if args.cli: + cli = Path(args.cli).resolve() + else: + env["GOBIN"] = str(directory / "bin") + run([args.go, "install", f"{MODULE}/cmd/go-omni@{args.version}"], directory, env) + cli = directory / "bin" / "go-omni" + command = [str(cli), "new", "--module", "example.com/databaseCheck", "--database", "mysql"] + if args.framework_dir: + command += ["--framework-dir", str(Path(args.framework_dir).resolve())] + project = directory / "billing" + run(command + [str(project)], directory, env) + module = json.loads(run([args.go, "mod", "edit", "-json"], project, env)) + if not args.framework_dir and module.get("Replace"): + raise AssertionError("remote MySQL project contains a local replace") + if {entry["Path"]: entry["Version"] for entry in module["Require"]}.get(MODULE) != args.version: + raise AssertionError("MySQL project did not pin the CLI version") + run([args.go, "mod", "tidy"], project, env) + binary = directory / "server" + run([args.go, "build", "-o", str(binary), "./cmd/server"], project, env) + database = MySQL(directory, env, args.image) + servers = [] + try: + database.start() + database.sql((project / "schema.sql").read_text(), "omni_test") + testEnv = dict(env, GO_OMNI_TEST_MYSQL_DSN=database.dsn) + output = run([args.go, "test", "-race", "-v", "./..."], project, testEnv) + for name in ("TestDatabaseTransactions", "TestDatabasePoolHealthAndClose"): + if f"--- PASS: {name}" not in output: + raise AssertionError(f"integration test did not pass: {name}\n{output}") + print("PASS real MySQL: commit, error/panic/cancel rollback, pool limits, health timeout, close", flush=True) + + serverEnv = dict(env, GO_OMNI_DB_DSN=database.dsn, GO_OMNI_MODE="release", + GO_OMNI_DB_PING_TIMEOUT="500ms", GO_OMNI_ENABLE_LOGGER="true") + server = Server(binary, project, serverEnv) + servers.append(server) + server.start() + ready = server.request("GET", "/health/ready") + if ready["code"] != "READY" or ready["data"]["checks"][0]["name"] != "database": + raise AssertionError(f"database probe not registered: {ready}") + server.request("POST", "/users", {"name": "", "email": "invalid"}, expected=400) + created = server.request("POST", "/users", {"name": " Alice ", "email": "ALICE@example.com"}, expected=201) + user = created["data"] + if user["email"] != "alice@example.com" or not user.get("createdAt"): + raise AssertionError(f"normalization/time parsing failed: {user}") + path = f"/users/{user['id']}" + duplicate = server.request("POST", "/users", {"name": "Duplicate", "email": "alice@example.com"}, expected=409) + if duplicate["code"] != "EMAIL_TAKEN": + raise AssertionError("duplicate email error was not mapped") + if server.request("GET", path)["data"]["name"] != "Alice": + raise AssertionError("GET did not load created user") + page = server.request("GET", "/users?page=1&pageSize=1")["data"] + if page["total"] != 1 or page["pageSize"] != 1: + raise AssertionError(f"pagination mismatch: {page}") + server.request("PUT", path, {"name": "Updated", "email": "updated@example.com"}) + server.stop() + waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0", + "connection pool closed on service exit") + server = Server(binary, project, serverEnv) + servers.append(server) + server.start() + if server.request("GET", path)["data"]["name"] != "Updated": + raise AssertionError("data did not survive service restart") + print("PASS HTTP validation, unique email, create/read/update/list, restart persistence and pool cleanup", flush=True) + + failureEnv = dict(serverEnv, GO_OMNI_DB_DSN=database.dsn.replace("/omni_test?", "/omni_empty?")) + failed = subprocess.run([str(binary)], cwd=project, env=failureEnv, capture_output=True, text=True, timeout=10) + if failed.returncode == 0 or "apply schema.sql" not in failed.stderr: + raise AssertionError("missing schema did not fail startup before listening") + waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE DB='omni_empty'") == "0", + "connection pool closed on schema failure") + print("PASS missing schema fails startup and releases connections", flush=True) + + run(["docker", "stop", "--time", "10", database.container], directory, database.env) + readiness = server.request("GET", "/health/ready", expected=503) + if readiness["code"] != "NOT_READY" or readiness["data"]["checks"][0]["status"] != "not_ready": + raise AssertionError("database outage did not fail readiness") + server.request("GET", "/health/live") + unavailable = server.request("GET", path, expected=500) + if unavailable["code"] != "INTERNAL_ERROR" or database.password in json.dumps(unavailable): + raise AssertionError("database outage leaked an internal error") + failed = subprocess.run([str(binary)], cwd=project, env=serverEnv, capture_output=True, text=True, timeout=10) + if failed.returncode == 0 or "ping database" not in failed.stderr: + raise AssertionError("offline database did not fail startup") + run(["docker", "start", database.container], directory, database.env) + database.wait() + def recovered(): + try: + return server.request("GET", "/health/ready")["code"] == "READY" + except (AssertionError, OSError, urllib.error.URLError): + return False + waitFor(recovered, "application reconnects after MySQL restart") + if server.request("GET", path)["data"]["name"] != "Updated": + raise AssertionError("database restart lost persistent data") + server.request("DELETE", path, expected=204) + server.request("GET", path, expected=404) + server.stop() + waitFor(lambda: database.sql("SELECT COUNT(*) FROM information_schema.processlist WHERE USER='omni_test'") == "0", + "final connection cleanup") + print("PASS database outage: readiness 503, liveness 200, startup failure; recovery 200; delete 204/404", flush=True) + finally: + for server in servers: + server.close() + database.close() + print("Removed test container, anonymous volume and temporary project", flush=True) + + +if __name__ == "__main__": + main() diff --git a/scripts/verifyScaffold.py b/scripts/verifyScaffold.py index 7939669..12ab4a9 100644 --- a/scripts/verifyScaffold.py +++ b/scripts/verifyScaffold.py @@ -65,7 +65,8 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None): status, body = request("/health/ready") if status != 200 or body["code"] != "READY": raise AssertionError(f"unexpected readiness: {body}") - status, user = request("/users", json.dumps({"name": "Smoke", "email": "smoke@example.com"}).encode()) + status, body = request("/users", json.dumps({"name": "Smoke", "email": "smoke@example.com"}).encode()) + user = body.get("data", body) if status != 201 or user["name"] != "Smoke" or "createdAt" not in user: raise AssertionError(f"unexpected created user: {user}") process.send_signal(stopSignal) @@ -90,7 +91,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None): def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--go", default="go", help="Go executable") - parser.add_argument("--version", default="v0.1.1", help="Released CLI version to install") + parser.add_argument("--version", default="v0.2.0", help="Released CLI version to install") parser.add_argument("--cli", help="Use a locally built CLI instead of installing") parser.add_argument("--framework-version", help="Explicit version passed to the CLI") parser.add_argument("--framework-dir", help="Local checkout for pre-release verification")