cryptoHermes/internal/repo/persistent/postgres/kline_repo.go

package postgres

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"

	"cryptoHermes/internal/entity"
)

type KlineRepo struct {
	pool *pgxpool.Pool
}

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