Veyra/internal/repository/market.go

package repository

import (
	"context"
	"encoding/json"
	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
	"git.sundynix.cn/Blizzard/Veyra/internal/model"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
	"time"
)

type MarketRepository struct{ pool *pgxpool.Pool }

func NewMarketRepository(pool *pgxpool.Pool) *MarketRepository { return &MarketRepository{pool} }
func (r *MarketRepository) List(ctx context.Context, query string) ([]model.Instrument, error) {
	rows, e := r.pool.Query(ctx, `SELECT id,symbol,exchange,name,price_cents,observed_at,received_at,source_id,source_version FROM veyra_instrument WHERE symbol LIKE $1 OR name LIKE $1 ORDER BY (price_cents>0) DESC,symbol LIMIT 100`, "%"+query+"%")
	if e != nil {
		return nil, mapError(e)
	}
	defer rows.Close()
	list := make([]model.Instrument, 0)
	for rows.Next() {
		var v model.Instrument
		if e := rows.Scan(&v.ID, &v.Symbol, &v.Exchange, &v.Name, &v.PriceCents, &v.ObservedAt, &v.ReceivedAt, &v.SourceID, &v.SourceVersion); e != nil {
			return nil, mapError(e)
		}
		v.Price = model.FormatMoney(v.PriceCents)
		list = append(list, v)
	}
	return list, mapError(rows.Err())
}
func (r *MarketRepository) Save(ctx context.Context, c model.ProviderConfig, list []model.Instrument, started time.Time, message string) error {
	tx, e := r.pool.Begin(ctx)
	if e != nil {
		return mapError(e)
	}
	defer tx.Rollback(context.Background())
	var version int64
	var enabled bool
	if e := tx.QueryRow(ctx, `SELECT version,enabled FROM veyra_provider_config WHERE id=$1 FOR SHARE`, c.ID).Scan(&version, &enabled); e != nil {
		return mapError(e)
	}
	if !enabled || version != c.Version {
		return apperror.ErrConflict
	}
	status := "COMPLETED"
	if message != "" {
		status = "FAILED"
		list = nil
	}
	for _, v := range list {
		_, e := tx.Exec(ctx, `INSERT INTO veyra_instrument(id,symbol,exchange,name,price_cents,observed_at,source_id,source_version) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(exchange,symbol) DO UPDATE SET name=excluded.name,price_cents=excluded.price_cents,observed_at=excluded.observed_at,received_at=now(),source_id=excluded.source_id,source_version=excluded.source_version WHERE veyra_instrument.observed_at<=excluded.observed_at`, uuid.NewString(), v.Symbol, v.Exchange, v.Name, v.PriceCents, v.ObservedAt, c.ID, c.Version)
		if e != nil {
			return mapError(e)
		}
	}
	if message == "" {
		// Quote events are produced only for tracked plans and truly recent vendor observations.
		// Re-reading an unchanged close cannot enqueue another price event.
		rows, err := tx.Query(ctx, `INSERT INTO veyra_quote_snapshot(id,symbol,exchange,data_mode,price_minor,observed_at,version,source_id,source_version)
 SELECT gen_random_uuid()::text,i.symbol,i.exchange,'public',i.price_cents,i.observed_at,1,i.source_id,i.source_version FROM veyra_instrument i
 WHERE i.source_id=$1 AND i.source_version=$2 AND i.observed_at>=now()-interval '1 minute' AND i.observed_at<=now()+interval '1 minute'
 AND EXISTS(SELECT 1 FROM veyra_trade_plan p JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_candidate c ON c.id=r.candidate_id WHERE `+activePlanPredicate+` AND r.snapshot->>'symbol'=i.symbol AND r.snapshot->>'exchange'=i.exchange)
 ON CONFLICT(exchange,symbol,data_mode) DO UPDATE SET price_minor=EXCLUDED.price_minor,observed_at=EXCLUDED.observed_at,version=veyra_quote_snapshot.version+1,source_id=EXCLUDED.source_id,source_version=EXCLUDED.source_version,updated_at=now()
 WHERE EXCLUDED.observed_at>veyra_quote_snapshot.observed_at RETURNING `+quoteColumns, c.ID, c.Version)
		if err != nil {
			return mapError(err)
		}
		quotes := make([]model.QuoteSnapshot, 0)
		for rows.Next() {
			q, err := scanQuote(rows)
			if err != nil {
				rows.Close()
				return err
			}
			quotes = append(quotes, q)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return mapError(err)
		}
		for _, q := range quotes {
			payload, err := json.Marshal(q)
			if err != nil {
				return apperror.ErrUnavailable
			}
			if _, err := tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,'quote_snapshot',$2,$3,'quote.observed',$4)`, uuid.NewString(), q.ID, q.Version, payload); err != nil {
				return mapError(err)
			}
		}
	}
	_, e = tx.Exec(ctx, `INSERT INTO veyra_market_sync(id,source_id,source_version,instrument_count,status,message,started_at) VALUES($1,$2,$3,$4,$5,$6,$7)`, uuid.NewString(), c.ID, c.Version, len(list), status, message, started)
	if e != nil {
		return mapError(e)
	}
	return mapError(tx.Commit(ctx))
}
func (r *MarketRepository) Tasks(ctx context.Context) ([]model.MarketSync, error) {
	rows, e := r.pool.Query(ctx, `SELECT id,source_id,source_version,instrument_count,status,message,started_at,finished_at FROM veyra_market_sync ORDER BY finished_at DESC,id DESC LIMIT 20`)
	if e != nil {
		return nil, mapError(e)
	}
	defer rows.Close()
	list := make([]model.MarketSync, 0)
	for rows.Next() {
		var v model.MarketSync
		if e := rows.Scan(&v.ID, &v.SourceID, &v.SourceVersion, &v.InstrumentCount, &v.Status, &v.Message, &v.StartedAt, &v.FinishedAt); e != nil {
			return nil, mapError(e)
		}
		list = append(list, v)
	}
	return list, mapError(rows.Err())
}