ee420f9c56
为什么用 pgxmock：testcontainers 需要 docker-in-CI 而当前还没有 CI； mock 路线 1) 零基础设施 2) 把 SQL 字符串和参数顺序作为契约锁死， schema 漂移立刻可见——这正是上生产前最需要的信号。pgxmock 升级到 testcontainers 不阻塞 v2，留 v2.1 做 nightly 集成测试。 关键改动： - internal/repo/persistent/postgres/pool.go：新建 pgxPool 接口， 列出 5 repo 实际用到的 Begin/Query/Exec 三个方法。生产代码继续 传 *pgxpool.Pool（自动满足接口），测试传 pgxmock.PgxPoolIface。 - 5 个 repo struct 字段从 *pgxpool.Pool 改为 pgxPool 接口；构造器 签名保留 *pgxpool.Pool，app.go DI 不动。 - 5 个 *_repo_test.go：每个 repo 至少 · UpsertMany happy path（精确 SQL + 参数） · UpsertMany 跳过逻辑（空字段 / 未收线） · UpsertMany ExecError（Rollback 路径） · FindRecent happy path（验证 DESC → 反序为升序） · FindRecent QueryError KlineRepo 额外加了 EmptyShortCircuits 和 BeginError 两个边界。 覆盖率：postgres 包 87.6%（UpsertMany ~86%、FindRecent ~93%）。 New*Repo 构造器为 0% 是设计选择：测试直接构造 struct 避开 *pgxpool.Pool 依赖。 所有 12 条守卫扫过无输出。
115 lines
2.8 KiB
Go
115 lines
2.8 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"cryptoHermes/internal/entity"
|
|
)
|
|
|
|
type KlineRepo struct {
|
|
pool pgxPool
|
|
}
|
|
|
|
func NewKlineRepo(pool *pgxpool.Pool) *KlineRepo {
|
|
return &KlineRepo{pool: pool}
|
|
}
|
|
|
|
const klineUpsertSQL = `
|
|
INSERT INTO market_klines (
|
|
source, symbol, interval, open_time, close_time,
|
|
open, high, low, close,
|
|
volume, quote_volume,
|
|
trade_count, taker_buy_base_volume, taker_buy_quote_volume,
|
|
updated_at
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5,
|
|
$6, $7, $8, $9,
|
|
$10, $11,
|
|
$12, $13, $14,
|
|
now()
|
|
)
|
|
ON CONFLICT (source, symbol, interval, open_time) DO UPDATE SET
|
|
close_time = EXCLUDED.close_time,
|
|
open = EXCLUDED.open,
|
|
high = EXCLUDED.high,
|
|
low = EXCLUDED.low,
|
|
close = EXCLUDED.close,
|
|
volume = EXCLUDED.volume,
|
|
quote_volume = EXCLUDED.quote_volume,
|
|
trade_count = EXCLUDED.trade_count,
|
|
taker_buy_base_volume = EXCLUDED.taker_buy_base_volume,
|
|
taker_buy_quote_volume = EXCLUDED.taker_buy_quote_volume,
|
|
updated_at = now()
|
|
`
|
|
|
|
func (r *KlineRepo) UpsertMany(ctx context.Context, items []entity.Kline) error {
|
|
if len(items) == 0 {
|
|
return nil
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("begin: %w", err)
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
for _, k := range items {
|
|
if !k.IsClosed {
|
|
continue
|
|
}
|
|
_, err := tx.Exec(ctx, klineUpsertSQL,
|
|
k.Source, k.Symbol, k.Interval, k.OpenTime, k.CloseTime,
|
|
k.Open, k.High, k.Low, k.Close,
|
|
k.Volume, k.QuoteVolume,
|
|
k.TradeCount, k.TakerBuyBaseVolume, k.TakerBuyQuoteVolume,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("upsert kline %s/%s@%d: %w", k.Symbol, k.Interval, k.OpenTime, err)
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
const klineFindSQL = `
|
|
SELECT source, symbol, interval, open_time, close_time,
|
|
open::text, high::text, low::text, close::text,
|
|
volume::text, quote_volume::text,
|
|
trade_count, taker_buy_base_volume::text, taker_buy_quote_volume::text
|
|
FROM market_klines
|
|
WHERE symbol = $1 AND interval = $2
|
|
ORDER BY open_time DESC
|
|
LIMIT $3
|
|
`
|
|
|
|
func (r *KlineRepo) FindRecent(ctx context.Context, symbol, interval string, limit int) ([]entity.Kline, error) {
|
|
if limit <= 0 {
|
|
limit = 300
|
|
}
|
|
rows, err := r.pool.Query(ctx, klineFindSQL, symbol, interval, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := make([]entity.Kline, 0, limit)
|
|
for rows.Next() {
|
|
var k entity.Kline
|
|
if err := rows.Scan(
|
|
&k.Source, &k.Symbol, &k.Interval, &k.OpenTime, &k.CloseTime,
|
|
&k.Open, &k.High, &k.Low, &k.Close,
|
|
&k.Volume, &k.QuoteVolume,
|
|
&k.TradeCount, &k.TakerBuyBaseVolume, &k.TakerBuyQuoteVolume,
|
|
); err != nil {
|
|
return nil, fmt.Errorf("scan: %w", err)
|
|
}
|
|
k.IsClosed = true
|
|
out = append(out, k)
|
|
}
|
|
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
|
|
out[i], out[j] = out[j], out[i]
|
|
}
|
|
return out, rows.Err()
|
|
}
|