Compare commits

..

1 Commits

Author SHA1 Message Date
affd02d5f7 feat: scaffold MySQL services with lifecycle and readiness checks 2026-09-20 23:12:47 +08:00
28 changed files with 1153 additions and 63 deletions

1
.gitignore vendored
View File

@@ -5,3 +5,4 @@ bin/
dist/ dist/
*.log *.log
coverage.out coverage.out
__pycache__/

View File

@@ -1,5 +1,14 @@
# Changelog # 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.1.1
- 修复脚手架生成项目固定依赖不存在的 `v0.0.0`;远程安装的 CLI 默认使用自身版本,支持 `--framework-version` 指定完整版本号。 - 修复脚手架生成项目固定依赖不存在的 `v0.0.0`;远程安装的 CLI 默认使用自身版本,支持 `--framework-version` 指定完整版本号。

View File

@@ -5,13 +5,13 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认用 Gi
远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本: 远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本:
```sh ```sh
go get git.yuntuo.hk/k2safe/go-omni@v0.1.1 go get git.yuntuo.hk/k2safe/go-omni@v0.2.0
``` ```
创建新业务项目: 创建新业务项目:
```sh ```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 go-omni new --module example.com/billing ./billing
cd billing cd billing
go mod tidy go mod tidy
@@ -21,6 +21,10 @@ go run ./cmd/server
远程安装的 CLI 默认引用自身版本,生成的项目不需要本地 `replace`。 远程安装的 CLI 默认引用自身版本,生成的项目不需要本地 `replace`。
具体版本和本地开发选项见 [CLI 文档](cmd/go-omni/README.md)。 具体版本和本地开发选项见 [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 ```text
@@ -148,7 +152,11 @@ application.GET("/health/ready", checker.ReadinessHandler())
新项目脚手架会生成项目自己的 `internal/servicecontext`。它集中保存 Logger、数据库、缓存和外部客户端,Service 只依赖这个项目上下文,不把基础设施初始化散落到各个 Handler: 新项目脚手架会生成项目自己的 `internal/servicecontext`。它集中保存 Logger、数据库、缓存和外部客户端,Service 只依赖这个项目上下文,不把基础设施初始化散落到各个 Handler:
```go ```go
services := servicecontext.New(logger, users) services, err := servicecontext.New(ctx, logger)
if err != nil {
return err
}
defer services.Close()
userService := service.NewUsers(services) userService := service.NewUsers(services)
``` ```

View File

@@ -8,11 +8,15 @@
- 自动验收入口:`python3 scripts/verifyScaffold.py --version v0.1.1`。 - 自动验收入口:`python3 scripts/verifyScaffold.py --version v0.1.1`。
- 已通过全量 race 测试、vet、build,以及从远程提交安装 CLI 后的无 replace 端到端验收;覆盖生成服务和 `App.Run` 的两种退出信号。 - 已通过全量 race 测试、vet、build,以及从远程提交安装 CLI 后的无 replace 端到端验收;覆盖生成服务和 `App.Run` 的两种退出信号。
## 2. 业务数据接入(待完成) ## 2. 业务数据接入(已完成,v0.2.0)
- 数据库初始化、Repository 和 ServiceContext 组合成完整业务启动流程。 - 数据库初始化、Repository 和 ServiceContext 组合成完整业务启动流程。
- Readiness 注册数据库探针,启动失败与服务退出时清理连接池。 - Readiness 注册数据库探针,启动失败与服务退出时清理连接池。
- 验收:真实数据库 CRUD、事务回滚、数据库不可用时 readiness 返回 503。 - 验收:真实数据库 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. 网络适配一致性(待完成) ## 3. 网络适配一致性(待完成)
@@ -25,4 +29,4 @@
- 在一个实际业务模块中使用远程固定版本,验证配置、日志、错误处理和升级流程。 - 在一个实际业务模块中使用远程固定版本,验证配置、日志、错误处理和升级流程。
- 根据真实接入问题收敛公共 API,再评估稳定版范围。 - 根据真实接入问题收敛公共 API,再评估稳定版范围。
以上阶段以验收结果判断完成。当前仍是基础框架试用阶段,数据库生产验收、自研网络内核与实际业务接入尚未完成。 以上阶段以验收结果判断完成。当前仍是基础框架试用阶段;MySQL 功能验收已完成,生产压测、自研网络内核与实际业务接入尚未完成。

View File

@@ -3,7 +3,7 @@
从远程安装,生成一个带 Handler / Service / Repository / Model 分层的服务骨架: 从远程安装,生成一个带 Handler / Service / Repository / Model 分层的服务骨架:
```sh ```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 go-omni new --module example.com/billing ./billing
cd billing cd billing
go mod tidy go mod tidy
@@ -16,7 +16,7 @@ go run ./cmd/server
需要选择其他已发布版本时,使用完整版本号: 需要选择其他已发布版本时,使用完整版本号:
```sh ```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` 等浮动引用不接受为固定版本。 `--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 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 的关闭函数注册到应用生命周期。 退出信号通过 `signal.NotifyContext` 传入 `RunContext`,Logger 的关闭函数注册到应用生命周期。
发布验收使用 Python 3 标准库脚本,不需要额外 Python 依赖: 发布验收使用 Python 3 标准库脚本,不需要额外 Python 依赖:
```sh ```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`,再下载依赖、 脚本在源码目录外安装远程 CLI、生成项目,确认固定版本且没有 `replace`,再下载依赖、
@@ -50,4 +61,7 @@ python3 scripts/verifyScaffold.py --version v0.1.1
```sh ```sh
go build -o /tmp/go-omni ./cmd/go-omni go build -o /tmp/go-omni ./cmd/go-omni
python3 scripts/verifyScaffold.py --cli /tmp/go-omni --framework-dir . python3 scripts/verifyScaffold.py --cli /tmp/go-omni --framework-dir .
python3 scripts/verifyDatabase.py --cli /tmp/go-omni --framework-dir .
``` ```
数据库验收需要本地 Docker,使用独立 MySQL 8.4 容器;不会连接已有业务数据库,结束后清理测试容器与数据卷。

View File

@@ -39,6 +39,17 @@ func runNew(args []string) error {
switch { switch {
case arg == "--force": case arg == "--force":
options.Force = true 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": case arg == "--module":
if index+1 >= len(args) { if index+1 >= len(args) {
return errors.New("--module requires a value") return errors.New("--module requires a value")
@@ -95,4 +106,5 @@ func usage() {
fmt.Fprintln(os.Stderr, " go-omni new [--module <module>] [--force] <directory>") fmt.Fprintln(os.Stderr, " go-omni new [--module <module>] [--force] <directory>")
fmt.Fprintln(os.Stderr, " --framework-version <version> Pin a remote framework version (defaults to CLI version)") fmt.Fprintln(os.Stderr, " --framework-version <version> Pin a remote framework version (defaults to CLI version)")
fmt.Fprintln(os.Stderr, " --framework-dir <directory> Use a local framework checkout") fmt.Fprintln(os.Stderr, " --framework-dir <directory> Use a local framework checkout")
fmt.Fprintln(os.Stderr, " --database <memory|mysql> Select generated storage (default memory)")
} }

View File

@@ -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)
}
}
}

View File

@@ -4,6 +4,7 @@ import (
"embed" "embed"
"errors" "errors"
"fmt" "fmt"
"go/format"
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
@@ -22,6 +23,7 @@ type Options struct {
Module string Module string
FrameworkDir string FrameworkDir string
FrameworkVersion string FrameworkVersion string
Database string
Force bool Force bool
} }
@@ -30,6 +32,8 @@ var files = map[string]string{
"README.md": "templates/README.md.tmpl", "README.md": "templates/README.md.tmpl",
"cmd/server/main.go": "templates/cmd-server-main.go.tmpl", "cmd/server/main.go": "templates/cmd-server-main.go.tmpl",
"internal/model/user.go": "templates/internal-model-user.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/repository/memory.go": "templates/internal-repository-memory.go.tmpl",
"internal/servicecontext/context.go": "templates/internal-servicecontext-context.go.tmpl", "internal/servicecontext/context.go": "templates/internal-servicecontext-context.go.tmpl",
"internal/service/user.go": "templates/internal-service-user.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) == "" { if strings.TrimSpace(options.Module) == "" {
return errors.New("module path is required") 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) version := strings.TrimSpace(options.FrameworkVersion)
if version == "" { if version == "" {
version = currentFrameworkVersion() version = currentFrameworkVersion()
@@ -72,17 +83,34 @@ func Generate(options Options) error {
Module: options.Module, Module: options.Module,
FrameworkVersion: version, FrameworkVersion: version,
FrameworkReplace: replace, FrameworkReplace: replace,
Database: database,
} }
paths := make([]string, 0, len(files)) projectFiles := make(map[string]string, len(files)+3)
for path := range files { 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) paths = append(paths, path)
} }
sort.Strings(paths) sort.Strings(paths)
for _, path := range paths { for _, path := range paths {
content, err := render(files[path], data) content, err := render(projectFiles[path], data)
if err != nil { if err != nil {
return fmt.Errorf("render %s: %w", path, err) 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)) target := filepath.Join(directory, filepath.FromSlash(path))
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return fmt.Errorf("create %s: %w", filepath.Dir(path), err) return fmt.Errorf("create %s: %w", filepath.Dir(path), err)
@@ -105,6 +133,7 @@ type templateData struct {
Module string Module string
FrameworkVersion string FrameworkVersion string
FrameworkReplace string FrameworkReplace string
Database string
} }
func render(name string, data templateData) (string, error) { func render(name string, data templateData) (string, error) {

View File

@@ -39,7 +39,7 @@ func TestGenerateCreatesServiceLayout(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("read generated server: %v", err) 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) { if !strings.Contains(string(mainContent), expected) {
t.Fatalf("generated server missing %q: %s", expected, mainContent) 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) { func TestFrameworkReplaceUsesCanonicalPaths(t *testing.T) {
root := frameworkRoot(t) root := frameworkRoot(t)
project := filepath.Join(t.TempDir(), "generated") project := filepath.Join(t.TempDir(), "generated")

View File

@@ -4,6 +4,26 @@ This service was generated with `go-omni new`.
Framework dependency: `git.yuntuo.hk/k2safe/go-omni {{.FrameworkVersion}}`. 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 ```sh
go mod tidy go mod tidy
go run ./cmd/server go run ./cmd/server
@@ -26,10 +46,59 @@ go build -o server ./cmd/server
./server ./server
``` ```
`GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。健康探针目前没有 `GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。
数据库依赖检查;接入数据库后需要向 Checker 注册探针。 {{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 ```text
persistence, then add it to `internal/servicecontext`. 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`。

View File

@@ -10,10 +10,8 @@ import (
omni "git.yuntuo.hk/k2safe/go-omni" omni "git.yuntuo.hk/k2safe/go-omni"
"git.yuntuo.hk/k2safe/go-omni/config" "git.yuntuo.hk/k2safe/go-omni/config"
"git.yuntuo.hk/k2safe/go-omni/health"
"git.yuntuo.hk/k2safe/go-omni/logx" "git.yuntuo.hk/k2safe/go-omni/logx"
"{{.Module}}/internal/handler" "{{.Module}}/internal/handler"
"{{.Module}}/internal/repository"
"{{.Module}}/internal/service" "{{.Module}}/internal/service"
"{{.Module}}/internal/servicecontext" "{{.Module}}/internal/servicecontext"
) )
@@ -43,19 +41,17 @@ func run() error {
// Access logs and business logs share the same configured sink. // Access logs and business logs share the same configured sink.
logx.SetDefault(logger) logx.SetDefault(logger)
users := repository.NewMemoryUsers() services, err := servicecontext.New(ctx, logger)
services := servicecontext.New(logger, users) if err != nil { return err }
defer services.Close()
userService := service.NewUsers(services) userService := service.NewUsers(services)
userHandler := handler.NewUsers(userService) userHandler := handler.NewUsers(userService)
checker, err := health.New()
if err != nil {
return err
}
application := omni.New(omni.WithConfig(settings)) application := omni.New(omni.WithConfig(settings))
application.OnShutdown(logger.Close) application.OnShutdown(logger.Close)
application.GET("/health/live", checker.LivenessHandler()) application.OnShutdown(services.Close)
application.GET("/health/ready", checker.ReadinessHandler()) application.GET("/health/live", services.Health.LivenessHandler())
application.GET("/health/ready", services.Health.ReadinessHandler())
userHandler.RegisterRoutes(application) userHandler.RegisterRoutes(application)
return application.RunContext(ctx, settings.Address) return application.RunContext(ctx, settings.Address)
} }

View File

@@ -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") }
}

View File

@@ -3,6 +3,8 @@ module {{.Module}}
go 1.22 go 1.22
require git.yuntuo.hk/k2safe/go-omni {{.FrameworkVersion}} 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}} {{- if .FrameworkReplace}}
replace git.yuntuo.hk/k2safe/go-omni => {{.FrameworkReplace}} replace git.yuntuo.hk/k2safe/go-omni => {{.FrameworkReplace}}
{{- end}} {{- end}}

View File

@@ -1,9 +1,10 @@
package handler package handler
import ( import (
"errors"
"net/http" "net/http"
omni "git.yuntuo.hk/k2safe/go-omni" "git.yuntuo.hk/k2safe/go-omni"
"{{.Module}}/internal/model" "{{.Module}}/internal/model"
"{{.Module}}/internal/service" "{{.Module}}/internal/service"
) )
@@ -12,20 +13,109 @@ type Users struct {
service *service.Users 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) { 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 { func (h *Users) Create(ctx *omni.Context) error {
var input model.UserInput var body userBody
if err := ctx.BindJSON(&input); err != nil { if err := ctx.BindJSON(&body); err != nil {
return err return err
} }
user, err := h.service.Create(ctx.Context(), input) user, err := h.service.Create(ctx.Request().Context(), model.UserInput(body))
if err != nil { 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 err
} }
return ctx.JSON(http.StatusCreated, user)
} }

View File

@@ -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")
)

View File

@@ -2,6 +2,7 @@ package model
import "time" import "time"
// User is the business entity; it has no dependency on HTTP or a database driver.
type User struct { type User struct {
ID int64 `json:"id"` ID int64 `json:"id"`
Name string `json:"name"` Name string `json:"name"`
@@ -10,7 +11,15 @@ type User struct {
UpdatedAt time.Time `json:"updatedAt"` UpdatedAt time.Time `json:"updatedAt"`
} }
// UserInput is shared by creation and full replacement (PUT).
type UserInput struct { type UserInput struct {
Name string `json:"name"` Name string
Email string `json:"email"` Email string
}
type UserPage struct {
Items []User `json:"items"`
Total int `json:"total"`
Page int `json:"page"`
PageSize int `json:"pageSize"`
} }

View File

@@ -2,30 +2,125 @@ package repository
import ( import (
"context" "context"
"sort"
"sync" "sync"
"time" "time"
"{{.Module}}/internal/model" "{{.Module}}/internal/model"
) )
type UserRepository interface { // MemoryUsers stores data only for this process. Replace it with a database
Create(context.Context, model.UserInput) (model.User, error) // implementation of UserRepository when persistence is required.
}
type MemoryUsers struct { type MemoryUsers struct {
mu sync.Mutex mu sync.RWMutex
users map[int64]model.User
emails map[string]int64
nextID 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() r.mu.Lock()
defer r.mu.Unlock() 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() 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.nextID++
r.users = append(r.users, user) r.users[user.ID] = user
r.emails[user.Email] = user.ID
return user, nil 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
}

View File

@@ -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
}

View File

@@ -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
}

View File

@@ -2,20 +2,61 @@ package service
import ( import (
"context" "context"
"net/mail"
"strings"
"unicode/utf8"
"{{.Module}}/internal/model" "{{.Module}}/internal/model"
"{{.Module}}/internal/servicecontext" "{{.Module}}/internal/servicecontext"
) )
type Users struct { type Users struct {
context *servicecontext.Context services *servicecontext.Context
} }
func NewUsers(context *servicecontext.Context) *Users { func NewUsers(services *servicecontext.Context) *Users {
return &Users{context: context} return &Users{services: services}
} }
func (s *Users) Create(ctx context.Context, input model.UserInput) (model.User, error) { func (s *Users) Create(ctx context.Context, input model.UserInput) (model.User, error) {
// Put business normalization and rules here. input, err := normalize(input)
return s.context.Users.Create(ctx, 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
} }

View File

@@ -1,6 +1,12 @@
package servicecontext package servicecontext
import ( 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" "git.yuntuo.hk/k2safe/go-omni/logx"
"{{.Module}}/internal/repository" "{{.Module}}/internal/repository"
) )
@@ -10,8 +16,51 @@ import (
type Context struct { type Context struct {
Logger *logx.Logger Logger *logx.Logger
Users repository.UserRepository Users repository.UserRepository
Health *health.Checker
{{if eq .Database "mysql"}}DB *database.DB{{end}}
} }
func New(logger *logx.Logger, users repository.UserRepository) *Context { // New initializes the project's dependencies before the HTTP listener opens.
return &Context{Logger: logger, Users: users} // 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
} }

View File

@@ -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;

View File

@@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni"
// Source builds can carry an unpublished VCS version (including +dirty). // Source builds can carry an unpublished VCS version (including +dirty).
// Keep this fallback in sync with the next release; installed CLIs use their // Keep this fallback in sync with the next release; installed CLIs use their
// downloaded module version, which also has a module checksum. // 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-]+)*)?$`) 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-]+)*)?$`)

View File

@@ -4,7 +4,7 @@
- `Config` / `ConfigFromEnv`:连接池参数和 `GO_OMNI_DB_*` 环境变量。 - `Config` / `ConfigFromEnv`:连接池参数和 `GO_OMNI_DB_*` 环境变量。
- `Open`:创建连接池、设置连接上限并在返回前执行 `PingContext`。 - `Open`:创建连接池、设置连接上限并在返回前执行 `PingContext`。
- `Health`:用于就绪检查。 - `Health`:用于就绪检查,等待连接和 Ping 均受 `PingTimeout` 限制。
- `InTx`:统一事务提交和回滚;工作函数返回错误或 panic 时回滚。 - `InTx`:统一事务提交和回滚;工作函数返回错误或 panic 时回滚。
- `Close`:由嵌入的 `*sql.DB` 提供,应用退出时显式调用。 - `Close`:由嵌入的 `*sql.DB` 提供,应用退出时显式调用。
@@ -18,7 +18,13 @@ import (
) )
cfg, err := database.ConfigFromEnv("GO_OMNI_DB_") cfg, err := database.ConfigFromEnv("GO_OMNI_DB_")
if err != nil {
return err
}
db, err := database.Open(context.Background(), cfg) db, err := database.Open(context.Background(), cfg)
if err != nil {
return err
}
defer db.Close() defer db.Close()
``` ```
@@ -35,3 +41,18 @@ GO_OMNI_DB_PING_TIMEOUT=5s
``` ```
`DSN` 必填,驱动默认是 `mysql`。数据库建库、迁移、表结构和具体 Repository 属于业务项目或可选模块,不会被核心包自动执行。 `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。

View File

@@ -93,6 +93,9 @@ type DB struct {
} }
func Open(ctx context.Context, cfg Config) (*DB, error) { func Open(ctx context.Context, cfg Config) (*DB, error) {
if ctx == nil {
ctx = context.Background()
}
if err := cfg.Validate(); err != nil { if err := cfg.Validate(); err != nil {
return nil, err return nil, err
} }
@@ -103,9 +106,9 @@ func Open(ctx context.Context, cfg Config) (*DB, error) {
if cfg.MaxOpenConns > 0 { if cfg.MaxOpenConns > 0 {
db.SetMaxOpenConns(cfg.MaxOpenConns) db.SetMaxOpenConns(cfg.MaxOpenConns)
} }
if cfg.MaxIdleConns > 0 { // database/sql uses zero to disable idle connections; do not leave its
db.SetMaxIdleConns(cfg.MaxIdleConns) // default pool size in place when the caller explicitly chooses zero.
} db.SetMaxIdleConns(cfg.MaxIdleConns)
db.SetConnMaxLifetime(cfg.ConnMaxLifetime) db.SetConnMaxLifetime(cfg.ConnMaxLifetime)
db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime) db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime)
checkCtx, cancel := context.WithTimeout(ctx, cfg.PingTimeout) 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 { if db == nil || db.DB == nil {
return errors.New("database is not initialized") 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. // InTx commits only when work returns nil. Any work error or panic rolls back.

View File

@@ -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)
}
}

265
scripts/verifyDatabase.py Normal file
View File

@@ -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()

View File

@@ -65,7 +65,8 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None):
status, body = request("/health/ready") status, body = request("/health/ready")
if status != 200 or body["code"] != "READY": if status != 200 or body["code"] != "READY":
raise AssertionError(f"unexpected readiness: {body}") 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: if status != 201 or user["name"] != "Smoke" or "createdAt" not in user:
raise AssertionError(f"unexpected created user: {user}") raise AssertionError(f"unexpected created user: {user}")
process.send_signal(stopSignal) process.send_signal(stopSignal)
@@ -90,7 +91,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None):
def main(): def main():
parser = argparse.ArgumentParser(description=__doc__) parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--go", default="go", help="Go executable") 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("--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-version", help="Explicit version passed to the CLI")
parser.add_argument("--framework-dir", help="Local checkout for pre-release verification") parser.add_argument("--framework-dir", help="Local checkout for pre-release verification")