You've already forked agentic-coding-workflow
feat(db): add pgxpool with health check and integration test
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// 默认连接池配置:10–50 用户私部署量级够用;超出后再按实际负载调整。
|
||||
const (
|
||||
defaultMaxConns = 20
|
||||
defaultMinConns = 2
|
||||
defaultMaxConnLifetime = time.Hour
|
||||
defaultPingTimeout = 5 * time.Second
|
||||
defaultHealthTimeout = time.Second
|
||||
)
|
||||
|
||||
// Pool 是 pgxpool.Pool 的类型别名,外部代码可直接以 db.Pool 表达连接池类型,
|
||||
// 同时保留 pgx 原生方法(Acquire/Exec/Query/...)。
|
||||
type Pool = pgxpool.Pool
|
||||
|
||||
// NewPool 根据 dsn 构建 Postgres 连接池,并在返回前用 5s 超时的 Ping 做健康验证。
|
||||
// 失败时关闭已创建的池并返回错误,避免泄露资源。
|
||||
func NewPool(ctx context.Context, dsn string) (*Pool, error) {
|
||||
cfg, err := pgxpool.ParseConfig(dsn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse dsn: %w", err)
|
||||
}
|
||||
cfg.MaxConns = defaultMaxConns
|
||||
cfg.MinConns = defaultMinConns
|
||||
cfg.MaxConnLifetime = defaultMaxConnLifetime
|
||||
|
||||
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("new pool: %w", err)
|
||||
}
|
||||
|
||||
pingCtx, cancel := context.WithTimeout(ctx, defaultPingTimeout)
|
||||
defer cancel()
|
||||
if err := pool.Ping(pingCtx); err != nil {
|
||||
pool.Close()
|
||||
return nil, fmt.Errorf("ping: %w", err)
|
||||
}
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
// HealthCheck 用 1s 超时探测数据库连通性,供 healthz/readyz 等探针调用。
|
||||
func HealthCheck(ctx context.Context, pool *Pool) error {
|
||||
c, cancel := context.WithTimeout(ctx, defaultHealthTimeout)
|
||||
defer cancel()
|
||||
return pool.Ping(c)
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
//go:build integration
|
||||
|
||||
package db_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
tcpg "github.com/testcontainers/testcontainers-go/modules/postgres"
|
||||
|
||||
mydb "github.com/yan1h/agent-coding-workflow/internal/infra/db"
|
||||
)
|
||||
|
||||
// TestNewPool_ConnectsAndPings 用 testcontainers 起一个真实的 Postgres,
|
||||
// 验证 NewPool 能连上、HealthCheck 能 Ping 通。
|
||||
// 仅在 integration build tag 下编译/运行;本地 `go test ./...` 会跳过。
|
||||
func TestNewPool_ConnectsAndPings(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// v0.42.0 起 RunContainer 已 Deprecated,统一用 Run(ctx, image, opts...)。
|
||||
container, err := tcpg.Run(ctx, "postgres:16-alpine",
|
||||
tcpg.WithDatabase("acw"),
|
||||
tcpg.WithUsername("acw"),
|
||||
tcpg.WithPassword("acw"),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = container.Terminate(ctx) })
|
||||
|
||||
dsn, err := container.ConnectionString(ctx, "sslmode=disable")
|
||||
require.NoError(t, err)
|
||||
|
||||
pool, err := mydb.NewPool(ctx, dsn)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(pool.Close)
|
||||
|
||||
hctx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
require.NoError(t, mydb.HealthCheck(hctx, pool))
|
||||
}
|
||||
Reference in New Issue
Block a user