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 条守卫扫过无输出。
91 lines
2.2 KiB
Go
91 lines
2.2 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"cryptoHermes/internal/entity"
|
|
)
|
|
|
|
type TakerVolumeRepo struct {
|
|
pool pgxPool
|
|
}
|
|
|
|
func NewTakerVolumeRepo(pool *pgxpool.Pool) *TakerVolumeRepo {
|
|
return &TakerVolumeRepo{pool: pool}
|
|
}
|
|
|
|
const takerUpsertSQL = `
|
|
INSERT INTO taker_buy_sell_volume (source, symbol, period, timestamp, buy_sell_ratio, buy_volume, sell_volume)
|
|
VALUES ($1, $2, $3, $4,
|
|
NULLIF($5, '')::NUMERIC,
|
|
NULLIF($6, '')::NUMERIC,
|
|
NULLIF($7, '')::NUMERIC)
|
|
ON CONFLICT (source, symbol, period, timestamp) DO UPDATE SET
|
|
buy_sell_ratio = EXCLUDED.buy_sell_ratio,
|
|
buy_volume = EXCLUDED.buy_volume,
|
|
sell_volume = EXCLUDED.sell_volume
|
|
`
|
|
|
|
func (r *TakerVolumeRepo) UpsertMany(ctx context.Context, items []entity.TakerBuySellVolume) error {
|
|
if len(items) == 0 {
|
|
return nil
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
for _, t := range items {
|
|
if t.Timestamp == 0 {
|
|
continue
|
|
}
|
|
if _, err := tx.Exec(ctx, takerUpsertSQL,
|
|
t.Source, t.Symbol, t.Period, t.Timestamp,
|
|
t.BuySellRatio, t.BuyVolume, t.SellVolume,
|
|
); err != nil {
|
|
return fmt.Errorf("upsert taker %s/%s@%d: %w", t.Symbol, t.Period, t.Timestamp, err)
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
const takerFindSQL = `
|
|
SELECT source, symbol, period, timestamp,
|
|
COALESCE(buy_sell_ratio::text, ''),
|
|
COALESCE(buy_volume::text, ''),
|
|
COALESCE(sell_volume::text, '')
|
|
FROM taker_buy_sell_volume
|
|
WHERE symbol = $1 AND period = $2
|
|
ORDER BY timestamp DESC
|
|
LIMIT $3
|
|
`
|
|
|
|
func (r *TakerVolumeRepo) FindRecent(ctx context.Context, symbol, period string, limit int) ([]entity.TakerBuySellVolume, error) {
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
rows, err := r.pool.Query(ctx, takerFindSQL, symbol, period, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := make([]entity.TakerBuySellVolume, 0, limit)
|
|
for rows.Next() {
|
|
var t entity.TakerBuySellVolume
|
|
if err := rows.Scan(&t.Source, &t.Symbol, &t.Period, &t.Timestamp,
|
|
&t.BuySellRatio, &t.BuyVolume, &t.SellVolume); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, t)
|
|
}
|
|
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()
|
|
}
|