Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| affd02d5f7 | |||
| ff95692edb | |||
| 0bd71b17b4 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -5,3 +5,4 @@ bin/
|
||||
dist/
|
||||
*.log
|
||||
coverage.out
|
||||
__pycache__/
|
||||
|
||||
23
CHANGELOG.md
Normal file
23
CHANGELOG.md
Normal file
@@ -0,0 +1,23 @@
|
||||
# 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` 指定完整版本号。
|
||||
- 显式远程版本禁用自动本地替换;本地替换路径支持空格,并按精确 module path 识别框架目录。
|
||||
- 生成项目通过 `signal.NotifyContext` 接入优雅退出,启动过程中的错误会先执行 defer 清理再报告。
|
||||
- `App.Run` 自动处理 `SIGINT` / `SIGTERM`;`RunContext` 仍由调用方控制,退出超时会关闭活跃连接。
|
||||
- 增加请求排空、超时取消、启动失败清理测试,以及远程脚手架和真实进程信号验收脚本。
|
||||
|
||||
## v0.1.0
|
||||
|
||||
- 首个可试用版本:应用生命周期、HTTP 抽象、Gin 与标准库路由适配器、统一响应与错误。
|
||||
- 提供配置、日志、校验、认证接口、数据库基础能力、健康探针和项目脚手架。
|
||||
34
README.md
34
README.md
@@ -5,9 +5,26 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认用 Gi
|
||||
远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本:
|
||||
|
||||
```sh
|
||||
go get git.yuntuo.hk/k2safe/go-omni@v0.1.0
|
||||
go get git.yuntuo.hk/k2safe/go-omni@v0.2.0
|
||||
```
|
||||
|
||||
创建新业务项目:
|
||||
|
||||
```sh
|
||||
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
|
||||
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
|
||||
@@ -25,7 +42,7 @@ go-omni/
|
||||
├── database/ # database/sql 生命周期和事务
|
||||
├── transport/gin/ # Gin 网络适配器,可替换
|
||||
├── examples/ # 示例项目
|
||||
└── cmd/ # 后续 CLI
|
||||
└── cmd/ # go-omni 项目脚手架
|
||||
```
|
||||
|
||||
## 快速开始
|
||||
@@ -104,7 +121,8 @@ application.OnShutdown(logger.Close)
|
||||
`GO_OMNI_ENVIRONMENT`、`GO_OMNI_HTTP_ADDR` 和 `GO_OMNI_SHUTDOWN_TIMEOUT`。数据库连接仍由
|
||||
`database.ConfigFromEnv` 从 `GO_OMNI_DB_*` 加载。
|
||||
|
||||
`App` 同时实现标准 `http.Handler`,并提供 `RunContext`:收到 `SIGINT` / `SIGTERM` 后可以先停止接收新请求,再在超时时间内等待活跃请求完成。数据库连接池由 `database.Open` 创建,在服务退出时调用 `Close`。
|
||||
`App` 同时实现标准 `http.Handler`。`Run` 监听 `SIGINT` / `SIGTERM`;`RunContext` 则由调用方的 context 控制退出,生成项目已使用 `signal.NotifyContext` 接好信号。
|
||||
退出时先停止接收新请求,等待活跃请求完成,再执行资源清理。超过 `ShutdownTimeout` 时强制关闭连接、取消请求 context,并返回超时错误。业务代码应响应 context 取消;框架无法强制结束忽略取消的业务 goroutine。数据库连接池由 `database.Open` 创建,在服务退出时调用 `Close`。
|
||||
|
||||
应用资源可以直接挂到 `omni.New` 返回的生命周期上:
|
||||
|
||||
@@ -134,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)
|
||||
```
|
||||
|
||||
@@ -154,4 +176,6 @@ go run ./examples/reference
|
||||
|
||||
## 当前边界
|
||||
|
||||
缓存、消息队列、定时任务和 CLI 暂时作为后续可选模块。认证层只定义 `Authenticator`、`Principal` 和路由保护中间件,具体 JWT、Session 或 OAuth 由项目实现。核心数据库层只依赖 `database/sql`,具体驱动、迁移和表结构由项目选择。
|
||||
缓存、消息队列和定时任务暂时作为后续可选模块。认证层只定义 `Authenticator`、`Principal` 和路由保护中间件,具体 JWT、Session 或 OAuth 由项目实现。核心数据库层只依赖 `database/sql`,具体驱动、迁移和表结构由项目选择。
|
||||
|
||||
当前成熟度与后续验收标准见 [ROADMAP.md](ROADMAP.md),版本变更见 [CHANGELOG.md](CHANGELOG.md)。
|
||||
|
||||
32
ROADMAP.md
Normal file
32
ROADMAP.md
Normal file
@@ -0,0 +1,32 @@
|
||||
# GoOmni 成熟度计划
|
||||
|
||||
## 1. 新项目启动与发布流程(已完成,v0.1.1)
|
||||
|
||||
- 修正脚手架固定版本引用,保留显式本地开发选项。
|
||||
- 生成服务处理退出信号,等待请求完成后清理资源。
|
||||
- 验收:源码目录外安装远程 CLI → 生成无 replace 的项目 → 编译 → 请求健康接口和业务接口 → SIGINT/SIGTERM 正常退出。
|
||||
- 自动验收入口:`python3 scripts/verifyScaffold.py --version v0.1.1`。
|
||||
- 已通过全量 race 测试、vet、build,以及从远程提交安装 CLI 后的无 replace 端到端验收;覆盖生成服务和 `App.Run` 的两种退出信号。
|
||||
|
||||
## 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. 网络适配一致性(待完成)
|
||||
|
||||
- Gin 与标准库适配器运行同一套契约测试。
|
||||
- 覆盖路由参数、分组、中间件、错误响应和取消传播。
|
||||
- 验收:切换适配器后,业务 Handler 和 Service 无需修改,约定的 HTTP 行为一致。
|
||||
|
||||
## 4. 实际业务试用(待完成)
|
||||
|
||||
- 在一个实际业务模块中使用远程固定版本,验证配置、日志、错误处理和升级流程。
|
||||
- 根据真实接入问题收敛公共 API,再评估稳定版范围。
|
||||
|
||||
以上阶段以验收结果判断完成。当前仍是基础框架试用阶段;MySQL 功能验收已完成,生产压测、自研网络内核与实际业务接入尚未完成。
|
||||
29
app/app.go
29
app/app.go
@@ -3,8 +3,12 @@ package app
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
@@ -104,9 +108,12 @@ func (a *App) Close() error {
|
||||
return a.closeErr
|
||||
}
|
||||
|
||||
// Run starts the HTTP server. The default address is :8080.
|
||||
// Run starts the HTTP server and handles SIGINT/SIGTERM. Use RunContext when
|
||||
// the caller owns cancellation. The default address is :8080.
|
||||
func (a *App) Run(address ...string) error {
|
||||
return a.RunContext(context.Background(), address...)
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
return a.RunContext(ctx, address...)
|
||||
}
|
||||
|
||||
// NewServer creates a standard HTTP server around the application.
|
||||
@@ -125,6 +132,9 @@ func (a *App) RunContext(ctx context.Context, address ...string) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return a.Close()
|
||||
}
|
||||
listenOn := a.config.Address
|
||||
if listenOn == "" {
|
||||
listenOn = ":8080"
|
||||
@@ -133,8 +143,16 @@ func (a *App) RunContext(ctx context.Context, address ...string) error {
|
||||
listenOn = address[0]
|
||||
}
|
||||
server := a.NewServer(listenOn)
|
||||
listener, err := net.Listen("tcp", server.Addr)
|
||||
if err != nil {
|
||||
return errors.Join(err, a.Close())
|
||||
}
|
||||
return a.serveContext(ctx, server, listener)
|
||||
}
|
||||
|
||||
func (a *App) serveContext(ctx context.Context, server *http.Server, listener net.Listener) error {
|
||||
serveErr := make(chan error, 1)
|
||||
go func() { serveErr <- server.ListenAndServe() }()
|
||||
go func() { serveErr <- server.Serve(listener) }()
|
||||
|
||||
select {
|
||||
case err := <-serveErr:
|
||||
@@ -151,6 +169,11 @@ func (a *App) RunContext(ctx context.Context, address ...string) error {
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancel()
|
||||
shutdownErr := server.Shutdown(shutdownCtx)
|
||||
if shutdownErr != nil {
|
||||
// Shutdown leaves active connections open on timeout. Close them so
|
||||
// request contexts are canceled before releasing shared resources.
|
||||
shutdownErr = errors.Join(shutdownErr, server.Close())
|
||||
}
|
||||
if err := <-serveErr; err != nil && err != http.ErrServerClosed {
|
||||
shutdownErr = errors.Join(shutdownErr, err)
|
||||
}
|
||||
|
||||
163
app/lifecycle_test.go
Normal file
163
app/lifecycle_test.go
Normal file
@@ -0,0 +1,163 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/transport/nethttp"
|
||||
)
|
||||
|
||||
type closingListener struct {
|
||||
net.Listener
|
||||
closed chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (l *closingListener) Close() error {
|
||||
err := l.Listener.Close()
|
||||
l.once.Do(func() { close(l.closed) })
|
||||
return err
|
||||
}
|
||||
|
||||
func startTestServer(t *testing.T, application *App) (*closingListener, context.CancelFunc, <-chan error) {
|
||||
t.Helper()
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wrapped := &closingListener{Listener: listener, closed: make(chan struct{})}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
t.Cleanup(func() { _ = listener.Close() })
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- application.serveContext(ctx, application.NewServer(""), wrapped) }()
|
||||
return wrapped, cancel, done
|
||||
}
|
||||
|
||||
func await(t *testing.T, event <-chan struct{}) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-event:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("timed out waiting for server event")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShutdownDrainsRequestBeforeCleanup(t *testing.T) {
|
||||
application := NewWithRouter(nethttp.New(), config.WithShutdownTimeout(3*time.Second))
|
||||
started, release, cleaned := make(chan struct{}), make(chan struct{}), make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
unblock := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(unblock)
|
||||
application.GET("/slow", func(ctx *httpx.Context) error {
|
||||
close(started)
|
||||
<-release
|
||||
return ctx.OK("finished")
|
||||
})
|
||||
application.OnShutdown(func() error { close(cleaned); return nil })
|
||||
listener, cancel, done := startTestServer(t, application)
|
||||
requestDone := make(chan error, 1)
|
||||
go func() {
|
||||
client := &http.Client{Timeout: 5 * time.Second}
|
||||
response, err := client.Get("http://" + listener.Addr().String() + "/slow")
|
||||
if err == nil {
|
||||
defer response.Body.Close()
|
||||
_, err = io.ReadAll(response.Body)
|
||||
if response.StatusCode != http.StatusOK {
|
||||
err = errors.New("request was not completed successfully")
|
||||
}
|
||||
}
|
||||
requestDone <- err
|
||||
}()
|
||||
await(t, started)
|
||||
cancel()
|
||||
await(t, listener.closed)
|
||||
select {
|
||||
case <-cleaned:
|
||||
t.Fatal("resources closed while a request was still running")
|
||||
default:
|
||||
}
|
||||
unblock()
|
||||
if err := <-requestDone; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("server did not finish shutdown")
|
||||
}
|
||||
await(t, cleaned)
|
||||
}
|
||||
|
||||
func TestShutdownTimeoutClosesActiveConnection(t *testing.T) {
|
||||
application := NewWithRouter(nethttp.New(), config.WithShutdownTimeout(25*time.Millisecond))
|
||||
started, canceled := make(chan struct{}), make(chan struct{})
|
||||
application.GET("/slow", func(ctx *httpx.Context) error {
|
||||
close(started)
|
||||
<-ctx.Context().Done()
|
||||
close(canceled)
|
||||
return nil
|
||||
})
|
||||
listener, cancel, done := startTestServer(t, application)
|
||||
requestDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(requestDone)
|
||||
client := &http.Client{Timeout: 5 * time.Second}
|
||||
response, err := client.Get("http://" + listener.Addr().String() + "/slow")
|
||||
if err == nil {
|
||||
response.Body.Close()
|
||||
}
|
||||
}()
|
||||
await(t, started)
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected shutdown deadline error, got %v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("shutdown ignored timeout")
|
||||
}
|
||||
await(t, canceled)
|
||||
await(t, requestDone)
|
||||
}
|
||||
|
||||
func TestRunContextStartupFailureAndCanceledContextCloseResources(t *testing.T) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer listener.Close()
|
||||
for _, canceled := range []bool{false, true} {
|
||||
application := NewWithRouter(nethttp.New())
|
||||
closeCount := 0
|
||||
application.OnShutdown(func() error { closeCount++; return nil })
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
if canceled {
|
||||
cancel()
|
||||
}
|
||||
err := application.RunContext(ctx, listener.Addr().String())
|
||||
cancel()
|
||||
if !canceled && err == nil {
|
||||
t.Fatal("expected occupied address error")
|
||||
}
|
||||
if canceled && err != nil {
|
||||
t.Fatalf("canceled start: %v", err)
|
||||
}
|
||||
_ = application.Close()
|
||||
if closeCount != 1 {
|
||||
t.Fatalf("cleanup ran %d times", closeCount)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,11 +1,67 @@
|
||||
# go-omni CLI
|
||||
|
||||
生成一个带 Handler / Service / Repository / Model 分层的服务骨架:
|
||||
从远程安装,生成一个带 Handler / Service / Repository / Model 分层的服务骨架:
|
||||
|
||||
```sh
|
||||
go run ./cmd/go-omni new --module example.com/billing ./billing
|
||||
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
|
||||
go run ./cmd/server
|
||||
```
|
||||
|
||||
通过 `go install ...@版本` 安装的 CLI 从构建信息中读取版本,并把它固定到生成项目的
|
||||
`go.mod`。不会使用不存在的 `v0.0.0`,也不会在普通业务目录中添加本地 `replace`。
|
||||
|
||||
需要选择其他已发布版本时,使用完整版本号:
|
||||
|
||||
```sh
|
||||
go-omni new --framework-version v0.2.0 --module example.com/billing ./billing
|
||||
```
|
||||
|
||||
`--framework-version` 禁用自动本地替换;`main`、`latest` 等浮动引用不接受为固定版本。
|
||||
源码构建的 CLI 使用代码中声明的目标发布版本;该版本未发布时请显式选择已发布版本,或用 `--framework-dir` 验证本地源码。
|
||||
|
||||
在 GoOmni 源码目录执行时,命令会自动在生成项目的 `go.mod` 中加入本地
|
||||
`replace git.yuntuo.hk/k2safe/go-omni => ...`,便于立即运行和测试。目标目录非空时命令会停止,明确传入
|
||||
`--force` 才会覆盖脚手架文件。
|
||||
|
||||
也可以在任意目录显式指定源码路径:
|
||||
|
||||
```sh
|
||||
go-omni new --framework-dir /path/to/go-omni --module example.com/billing ./billing
|
||||
```
|
||||
|
||||
默认生成内存存储版本;选择 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.2.0
|
||||
python3 scripts/verifyDatabase.py --version v0.2.0
|
||||
```
|
||||
|
||||
脚本在源码目录外安装远程 CLI、生成项目,确认固定版本且没有 `replace`,再下载依赖、
|
||||
编译、请求接口,并对真实二进制分别发送 `SIGINT` 和 `SIGTERM`。同时验证框架 `Run` 的
|
||||
信号处理和清理回调。每个进程都必须退出码为 0、释放监听端口;临时项目和测试进程会被清理。
|
||||
|
||||
发布前验证当前源码:
|
||||
|
||||
```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 容器;不会连接已有业务数据库,结束后清理测试容器与数据卷。
|
||||
|
||||
@@ -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")
|
||||
@@ -55,6 +66,17 @@ func runNew(args []string) error {
|
||||
options.FrameworkDir = args[index]
|
||||
case strings.HasPrefix(arg, "--framework-dir="):
|
||||
options.FrameworkDir = strings.TrimPrefix(arg, "--framework-dir=")
|
||||
case arg == "--framework-version":
|
||||
if index+1 >= len(args) || strings.TrimSpace(args[index+1]) == "" {
|
||||
return errors.New("--framework-version requires a value")
|
||||
}
|
||||
index++
|
||||
options.FrameworkVersion = args[index]
|
||||
case strings.HasPrefix(arg, "--framework-version="):
|
||||
options.FrameworkVersion = strings.TrimPrefix(arg, "--framework-version=")
|
||||
if strings.TrimSpace(options.FrameworkVersion) == "" {
|
||||
return errors.New("--framework-version requires a value")
|
||||
}
|
||||
case strings.HasPrefix(arg, "-"):
|
||||
return fmt.Errorf("unknown option %q", arg)
|
||||
case target == "":
|
||||
@@ -82,4 +104,7 @@ func usage() {
|
||||
fmt.Fprintln(os.Stderr, "")
|
||||
fmt.Fprintln(os.Stderr, "Usage:")
|
||||
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-dir <directory> Use a local framework checkout")
|
||||
fmt.Fprintln(os.Stderr, " --database <memory|mysql> Select generated storage (default memory)")
|
||||
}
|
||||
|
||||
51
cmd/go-omni/main_test.go
Normal file
51
cmd/go-omni/main_test.go
Normal file
@@ -0,0 +1,51 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestNewWithRemoteFrameworkVersion(t *testing.T) {
|
||||
for _, flag := range [][]string{{"--framework-version", "v0.1.0"}, {"--framework-version=v0.1.0"}} {
|
||||
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)
|
||||
}
|
||||
content, err := os.ReadFile(filepath.Join(project, "go.mod"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(string(content), "go-omni v0.1.0") || strings.Contains(string(content), "replace ") {
|
||||
t.Fatalf("CLI did not pin remote dependency: %s", content)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewRejectsEmptyFrameworkVersion(t *testing.T) {
|
||||
for _, args := range [][]string{{"--framework-version"}, {"--framework-version="}, {"--framework-version", " "}} {
|
||||
if err := runNew(args); err == nil {
|
||||
t.Fatalf("accepted missing framework version: %v", args)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,10 +4,12 @@ import (
|
||||
"embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
"go/format"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"text/template"
|
||||
)
|
||||
@@ -17,10 +19,12 @@ var templateFiles embed.FS
|
||||
|
||||
// Options controls project generation.
|
||||
type Options struct {
|
||||
Dir string
|
||||
Module string
|
||||
FrameworkDir string
|
||||
Force bool
|
||||
Dir string
|
||||
Module string
|
||||
FrameworkDir string
|
||||
FrameworkVersion string
|
||||
Database string
|
||||
Force bool
|
||||
}
|
||||
|
||||
var files = map[string]string{
|
||||
@@ -28,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",
|
||||
@@ -42,6 +48,20 @@ 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()
|
||||
}
|
||||
if !validFrameworkVersion(version) {
|
||||
return fmt.Errorf("framework version %q must be a complete v0 or v1 semantic version", version)
|
||||
}
|
||||
|
||||
directory, err := filepath.Abs(options.Dir)
|
||||
if err != nil {
|
||||
@@ -54,20 +74,43 @@ func Generate(options Options) error {
|
||||
return fmt.Errorf("create project directory: %w", err)
|
||||
}
|
||||
|
||||
// An explicit remote version opts out of automatic local development mode.
|
||||
replace := ""
|
||||
if options.FrameworkDir != "" || options.FrameworkVersion == "" {
|
||||
replace = frameworkReplace(directory, options.FrameworkDir)
|
||||
}
|
||||
data := templateData{
|
||||
Module: options.Module,
|
||||
FrameworkReplace: frameworkReplace(directory, options.FrameworkDir),
|
||||
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)
|
||||
@@ -88,7 +131,9 @@ func Generate(options Options) error {
|
||||
|
||||
type templateData struct {
|
||||
Module string
|
||||
FrameworkVersion string
|
||||
FrameworkReplace string
|
||||
Database string
|
||||
}
|
||||
|
||||
func render(name string, data templateData) (string, error) {
|
||||
@@ -149,7 +194,7 @@ func frameworkReplace(projectDir, configured string) string {
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return filepath.ToSlash(relative)
|
||||
return strconv.Quote(filepath.ToSlash(relative))
|
||||
}
|
||||
|
||||
func canonicalPath(path string) string {
|
||||
@@ -172,8 +217,14 @@ func findFrameworkRoot() string {
|
||||
for {
|
||||
moduleFile := filepath.Join(directory, "go.mod")
|
||||
content, readErr := os.ReadFile(moduleFile)
|
||||
if readErr == nil && strings.Contains(string(content), "module git.yuntuo.hk/k2safe/go-omni") {
|
||||
return directory
|
||||
if readErr == nil {
|
||||
for _, line := range strings.Split(string(content), "\n") {
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) >= 2 && fields[0] == "module" && strings.Trim(fields[1], "\"") == frameworkModule {
|
||||
return directory
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
parent := filepath.Dir(directory)
|
||||
if parent == directory {
|
||||
|
||||
@@ -3,6 +3,8 @@ package scaffold
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime/debug"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
@@ -37,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.Run(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)
|
||||
}
|
||||
@@ -54,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")
|
||||
@@ -73,11 +104,59 @@ func TestFrameworkReplaceUsesCanonicalPaths(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if replace != filepath.ToSlash(expected) {
|
||||
if replace != strconv.Quote(filepath.ToSlash(expected)) {
|
||||
t.Fatalf("expected canonical replace %q, got %q", filepath.ToSlash(expected), replace)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateRemoteProjectPinsVersion(t *testing.T) {
|
||||
for _, version := range []string{"v0.1.0", "v1.2.3-rc.1", "v0.0.0-20260920140624-8b19aa79eea0"} {
|
||||
t.Run(version, func(t *testing.T) {
|
||||
project := filepath.Join(t.TempDir(), "billing")
|
||||
if err := Generate(Options{Dir: project, Module: "example.com/billing", FrameworkVersion: version}); 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), "require "+frameworkModule+" "+version+"\n") || strings.Contains(string(content), "replace ") {
|
||||
t.Fatalf("expected pinned remote dependency without replace: %s", content)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateRejectsInvalidVersionBeforeWriting(t *testing.T) {
|
||||
for _, version := range []string{"main", "latest", "v1.2", "v01.2.3", "v2.0.0", "v0.1.0-01", "v0.1.0\nreplace bad => ./bad"} {
|
||||
project := filepath.Join(t.TempDir(), "billing")
|
||||
if err := Generate(Options{Dir: project, Module: "example.com/billing", FrameworkVersion: version}); err == nil {
|
||||
t.Fatalf("accepted invalid version %q", version)
|
||||
}
|
||||
if _, err := os.Stat(project); !os.IsNotExist(err) {
|
||||
t.Fatalf("invalid version %q created a project: %v", version, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFrameworkVersionUsesInstalledModuleMetadata(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
info *debug.BuildInfo
|
||||
want string
|
||||
}{
|
||||
{nil, defaultFrameworkVersion},
|
||||
{&debug.BuildInfo{Main: debug.Module{Path: frameworkModule, Version: "(devel)"}}, defaultFrameworkVersion},
|
||||
{&debug.BuildInfo{Main: debug.Module{Path: frameworkModule, Version: "v0.1.0+dirty"}}, defaultFrameworkVersion},
|
||||
{&debug.BuildInfo{Main: debug.Module{Path: "example.com/unrelated", Version: "v1.2.3"}}, defaultFrameworkVersion},
|
||||
{&debug.BuildInfo{Main: debug.Module{Path: frameworkModule, Version: "v0.9.2", Sum: "h1:test"}}, "v0.9.2"},
|
||||
{&debug.BuildInfo{Main: debug.Module{Path: frameworkModule, Version: "v0.0.0-20260920140624-8b19aa79eea0", Sum: "h1:test"}}, "v0.0.0-20260920140624-8b19aa79eea0"},
|
||||
} {
|
||||
if got := frameworkVersion(test.info); got != test.want {
|
||||
t.Fatalf("version = %q, want %q", got, test.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func frameworkRoot(t *testing.T) string {
|
||||
t.Helper()
|
||||
directory, err := os.Getwd()
|
||||
|
||||
@@ -2,6 +2,28 @@
|
||||
|
||||
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
|
||||
@@ -16,7 +38,67 @@ GO_OMNI_ENABLE_LOGGER=true GO_OMNI_LOG_LEVEL=debug go run ./cmd/server
|
||||
`GO_OMNI_HTTP_ADDR` 可以修改监听地址。生成项目默认提供 `/health/live` 和
|
||||
`/health/ready`,日志、请求 ID 和优雅退出等基础能力由 GoOmni 统一初始化;业务依赖集中放在项目自己的 `internal/servicecontext`。
|
||||
|
||||
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`.
|
||||
服务使用 `signal.NotifyContext` 处理 `SIGINT` / `SIGTERM`。部署时构建并运行服务二进制,
|
||||
使信号直接到达服务进程:
|
||||
|
||||
```sh
|
||||
go build -o server ./cmd/server
|
||||
./server
|
||||
```
|
||||
|
||||
`GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。
|
||||
{{if eq .Database "mysql"}}
|
||||
`/health/ready` 已注册数据库探针:连接失败返回 `503`,数据库恢复后重新返回 `200`。
|
||||
`/health/live` 始终只检查服务存活。`GO_OMNI_DB_PING_TIMEOUT` 限制初始化检查和数据库探针的等待时间。
|
||||
退出时先排空 HTTP 请求,再关闭数据库连接池,最后关闭日志。
|
||||
|
||||
数据库连接池参数:
|
||||
|
||||
```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`。
|
||||
|
||||
@@ -1,49 +1,57 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if err := run(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func run() error {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
settings, err := config.LoadFromEnv("")
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if strings.TrimSpace(settings.Log.Service) == "" {
|
||||
settings.Log.Service = "{{.Module}}"
|
||||
}
|
||||
logger, err := logx.New(settings.Log)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
defer logger.Close()
|
||||
// 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 {
|
||||
log.Fatal(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)
|
||||
if err := application.Run(settings.Address); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
return application.RunContext(ctx, settings.Address)
|
||||
}
|
||||
|
||||
106
cmd/go-omni/scaffold/templates/databaseIntegration_test.go.tmpl
Normal file
106
cmd/go-omni/scaffold/templates/databaseIntegration_test.go.tmpl
Normal 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") }
|
||||
}
|
||||
@@ -2,7 +2,9 @@ module {{.Module}}
|
||||
|
||||
go 1.22
|
||||
|
||||
require git.yuntuo.hk/k2safe/go-omni v0.0.0
|
||||
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}}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
11
cmd/go-omni/scaffold/templates/internal-model-errors.go.tmpl
Normal file
11
cmd/go-omni/scaffold/templates/internal-model-errors.go.tmpl
Normal 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")
|
||||
)
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
139
cmd/go-omni/scaffold/templates/internal-repository-mysql.go.tmpl
Normal file
139
cmd/go-omni/scaffold/templates/internal-repository-mysql.go.tmpl
Normal 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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
9
cmd/go-omni/scaffold/templates/schema.sql.tmpl
Normal file
9
cmd/go-omni/scaffold/templates/schema.sql.tmpl
Normal 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;
|
||||
31
cmd/go-omni/scaffold/version.go
Normal file
31
cmd/go-omni/scaffold/version.go
Normal file
@@ -0,0 +1,31 @@
|
||||
package scaffold
|
||||
|
||||
import (
|
||||
"regexp"
|
||||
"runtime/debug"
|
||||
)
|
||||
|
||||
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.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-]+)*)?$`)
|
||||
|
||||
func validFrameworkVersion(version string) bool {
|
||||
return frameworkVersionPattern.MatchString(version)
|
||||
}
|
||||
|
||||
func currentFrameworkVersion() string {
|
||||
info, _ := debug.ReadBuildInfo()
|
||||
return frameworkVersion(info)
|
||||
}
|
||||
|
||||
func frameworkVersion(info *debug.BuildInfo) string {
|
||||
if info != nil && info.Main.Path == frameworkModule && info.Main.Sum != "" && validFrameworkVersion(info.Main.Version) {
|
||||
return info.Main.Version
|
||||
}
|
||||
return defaultFrameworkVersion
|
||||
}
|
||||
@@ -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。
|
||||
|
||||
@@ -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.
|
||||
|
||||
39
database/lifecycle_test.go
Normal file
39
database/lifecycle_test.go
Normal 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
265
scripts/verifyDatabase.py
Normal 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()
|
||||
159
scripts/verifyScaffold.py
Normal file
159
scripts/verifyScaffold.py
Normal file
@@ -0,0 +1,159 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verify a generated service outside the source tree using a real server process."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
MODULE = "git.yuntuo.hk/k2safe/go-omni"
|
||||
|
||||
|
||||
def run(args, cwd, env):
|
||||
result = subprocess.run(args, cwd=cwd, env=env, text=True,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=180)
|
||||
if result.returncode:
|
||||
raise RuntimeError(f"{args[0]} {args[1:]} failed:\n{result.stdout}")
|
||||
return result.stdout
|
||||
|
||||
|
||||
def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None):
|
||||
with socket.socket() as listener:
|
||||
listener.bind(("127.0.0.1", 0))
|
||||
address = f"127.0.0.1:{listener.getsockname()[1]}"
|
||||
serverEnv = {key: value for key, value in env.items() if not key.startswith("GO_OMNI_")}
|
||||
serverEnv.update(GO_OMNI_HTTP_ADDR=address, GO_OMNI_MODE="release",
|
||||
GO_OMNI_ENABLE_LOGGER="true", GO_OMNI_SHUTDOWN_TIMEOUT="2s")
|
||||
if cleanupMarker:
|
||||
serverEnv["GO_OMNI_CLEANUP_MARKER"] = str(cleanupMarker)
|
||||
logFile = directory / f"{binary.name}-{stopSignal.name}.log"
|
||||
client = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||||
|
||||
def request(path, data=None):
|
||||
req = urllib.request.Request(f"http://{address}{path}", data=data,
|
||||
headers={"Content-Type": "application/json", "X-Request-ID": "smokeRequest"})
|
||||
with client.open(req, timeout=2) as response:
|
||||
if response.headers.get("X-Request-ID") != "smokeRequest":
|
||||
raise AssertionError("request ID was not preserved")
|
||||
return response.status, json.load(response)
|
||||
|
||||
with logFile.open("w") as logs:
|
||||
process = subprocess.Popen([str(binary)], cwd=directory, env=serverEnv,
|
||||
stdout=logs, stderr=subprocess.STDOUT)
|
||||
try:
|
||||
deadline = time.monotonic() + 15
|
||||
while True:
|
||||
if process.poll() is not None:
|
||||
raise RuntimeError(f"server exited during startup: {logFile.read_text()}")
|
||||
try:
|
||||
status, body = request("/health/live")
|
||||
break
|
||||
except (OSError, urllib.error.URLError):
|
||||
if time.monotonic() >= deadline:
|
||||
raise RuntimeError(f"server did not start: {logFile.read_text()}")
|
||||
time.sleep(0.05)
|
||||
if status != 200 or body["data"]["status"] != "ok":
|
||||
raise AssertionError(f"unexpected liveness: {body}")
|
||||
if not cleanupMarker:
|
||||
status, body = request("/health/ready")
|
||||
if status != 200 or body["code"] != "READY":
|
||||
raise AssertionError(f"unexpected readiness: {body}")
|
||||
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)
|
||||
exitCode = process.wait(timeout=8)
|
||||
if exitCode != 0:
|
||||
raise AssertionError(f"{stopSignal.name} exit code {exitCode}: {logFile.read_text()}")
|
||||
if cleanupMarker and cleanupMarker.read_text() != "closed":
|
||||
raise AssertionError("shutdown callback was not executed")
|
||||
# A built binary, rather than `go run`, lets us verify the actual
|
||||
# service exited without leaving its child listening in the background.
|
||||
with socket.socket() as connection:
|
||||
connection.settimeout(1)
|
||||
if connection.connect_ex(("127.0.0.1", int(address.rsplit(":", 1)[1]))) == 0:
|
||||
raise AssertionError("server still accepts connections after exit")
|
||||
finally:
|
||||
if process.poll() is None:
|
||||
process.kill()
|
||||
process.wait(timeout=5)
|
||||
print(f"PASS {binary.name}: HTTP checks, {stopSignal.name}, exit 0, listener closed", flush=True)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--go", default="go", help="Go executable")
|
||||
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")
|
||||
parser.add_argument("--skip-run-check", action="store_true", help="Skip core Run signal checks when testing an older framework")
|
||||
args = parser.parse_args()
|
||||
env = dict(os.environ, GOWORK="off")
|
||||
env["GOPRIVATE"] = ",".join(filter(None, [env.get("GOPRIVATE", ""), "git.yuntuo.hk"]))
|
||||
with tempfile.TemporaryDirectory(prefix="go omni smoke ") as temporary:
|
||||
directory = Path(temporary).resolve()
|
||||
if args.cli:
|
||||
cli = Path(args.cli).resolve()
|
||||
else:
|
||||
env["GOBIN"] = str(directory / "bin")
|
||||
print(f"Installing remote CLI {args.version}", flush=True)
|
||||
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/scaffoldSmoke"]
|
||||
if args.framework_version:
|
||||
command += ["--framework-version", args.framework_version]
|
||||
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))
|
||||
requirements = {entry["Path"]: entry["Version"] for entry in module["Require"]}
|
||||
expectedVersion = args.framework_version or args.version
|
||||
if requirements.get(MODULE) != expectedVersion:
|
||||
raise AssertionError(f"expected framework {expectedVersion}, got {requirements}")
|
||||
if not args.framework_dir and module.get("Replace"):
|
||||
raise AssertionError("remote generation must not contain local replace directives")
|
||||
print(f"PASS generated module pins {expectedVersion}; local replace: {bool(module.get('Replace'))}", flush=True)
|
||||
run([args.go, "mod", "tidy"], project, env)
|
||||
run([args.go, "test", "./..."], project, env)
|
||||
binary = directory / "server"
|
||||
run([args.go, "build", "-o", str(binary), "./cmd/server"], project, env)
|
||||
for stopSignal in (signal.SIGINT, signal.SIGTERM):
|
||||
exerciseServer(binary, project, env, stopSignal)
|
||||
|
||||
if not args.skip_run_check:
|
||||
# Verify the framework's convenient Run API independently from the
|
||||
# generated main's caller-owned signal context.
|
||||
helper = project / "cmd" / "runCheck"
|
||||
helper.mkdir()
|
||||
helper.joinpath("main.go").write_text('''package main
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
omni "git.yuntuo.hk/k2safe/go-omni"
|
||||
)
|
||||
func main() {
|
||||
app := omni.New(omni.WithMode("release"))
|
||||
app.GET("/health/live", func(ctx *omni.Context) error { return ctx.OK(map[string]string{"status": "ok"}) })
|
||||
app.OnShutdown(func() error { return os.WriteFile(os.Getenv("GO_OMNI_CLEANUP_MARKER"), []byte("closed"), 0600) })
|
||||
if err := app.Run(os.Getenv("GO_OMNI_HTTP_ADDR")); err != nil { log.Fatal(err) }
|
||||
}
|
||||
''')
|
||||
runBinary = directory / "runCheck"
|
||||
run([args.go, "build", "-o", str(runBinary), "./cmd/runCheck"], project, env)
|
||||
for stopSignal in (signal.SIGINT, signal.SIGTERM):
|
||||
marker = project / f"cleanup-{stopSignal.name}"
|
||||
exerciseServer(runBinary, project, env, stopSignal, marker)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user