Veyra/internal/repository/strategy.go

package repository

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

type StrategyRepository struct{ pool *pgxpool.Pool }

func NewStrategyRepository(pool *pgxpool.Pool) *StrategyRepository { return &StrategyRepository{pool} }

const strategyColumns = `id,code,name,status,latest_revision,active_version_id,version,created_at,updated_at`

func scanStrategy(row pgx.Row) (model.Strategy, error) {
	var s model.Strategy
	err := row.Scan(&s.ID, &s.Code, &s.Name, &s.Status, &s.LatestRevision, &s.ActiveVersionID, &s.Version, &s.CreatedAt, &s.UpdatedAt)
	return s, mapError(err)
}
func (r *StrategyRepository) List(ctx context.Context) ([]model.Strategy, error) {
	rows, err := r.pool.Query(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy ORDER BY updated_at DESC,id DESC LIMIT 100`)
	if err != nil {
		return nil, mapError(err)
	}
	defer rows.Close()
	list := make([]model.Strategy, 0)
	for rows.Next() {
		s, err := scanStrategy(rows)
		if err != nil {
			return nil, err
		}
		list = append(list, s)
	}
	return list, mapError(rows.Err())
}
func (r *StrategyRepository) Detail(ctx context.Context, id string) (model.StrategyDetail, error) {
	tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly})
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	defer tx.Rollback(context.Background())
	return strategyDetail(ctx, tx, id)
}
func strategyDetail(ctx context.Context, tx pgx.Tx, id string) (model.StrategyDetail, error) {
	s, err := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1`, id))
	detail := model.StrategyDetail{Strategy: s, Versions: make([]model.StrategyVersion, 0)}
	if err != nil {
		return detail, err
	}
	rows, err := tx.Query(ctx, `SELECT id,strategy_id,revision,algorithm,parameters,description,author_id,created_at FROM veyra_strategy_version WHERE strategy_id=$1 ORDER BY revision DESC`, id)
	if err != nil {
		return detail, mapError(err)
	}
	defer rows.Close()
	for rows.Next() {
		var v model.StrategyVersion
		var parameters []byte
		if err := rows.Scan(&v.ID, &v.StrategyID, &v.Revision, &v.Algorithm, &parameters, &v.Description, &v.AuthorID, &v.CreatedAt); err != nil {
			return detail, mapError(err)
		}
		if json.Unmarshal(parameters, &v.Parameters) != nil {
			return detail, apperror.ErrUnavailable
		}
		detail.Versions = append(detail.Versions, v)
	}
	return detail, mapError(rows.Err())
}
func (r *StrategyRepository) Save(ctx context.Context, id string, input service.StrategyInput, actor, requestID string) (model.StrategyDetail, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	defer tx.Rollback(context.Background())
	revision := 1
	if id == "" {
		id = uuid.NewString()
		_, err = tx.Exec(ctx, `INSERT INTO veyra_strategy(id,code,name) VALUES($1,$2,$3)`, id, input.Code, input.Name)
	} else {
		s, e := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1 FOR UPDATE`, id))
		if e != nil {
			return model.StrategyDetail{}, e
		}
		if s.Version != input.ExpectedVersion || s.Code != input.Code {
			return model.StrategyDetail{}, apperror.ErrConflict
		}
		revision = s.LatestRevision + 1
		_, err = tx.Exec(ctx, `UPDATE veyra_strategy SET name=$2,latest_revision=$3,version=version+1,updated_at=now() WHERE id=$1`, id, input.Name, revision)
	}
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	parameters, err := json.Marshal(input.Parameters)
	if err != nil {
		return model.StrategyDetail{}, apperror.ErrInvalidInput
	}
	versionID := uuid.NewString()
	_, err = tx.Exec(ctx, `INSERT INTO veyra_strategy_version(id,strategy_id,revision,algorithm,parameters,description,author_id) VALUES($1,$2,$3,'sma_cross',$4,$5,$6)`, versionID, id, revision, parameters, input.Description, actor)
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	if err := appendAudit(ctx, tx, actor, "STRATEGY_VERSION_CREATED", "strategy", id, requestID, map[string]any{"revision": revision, "version_id": versionID}); err != nil {
		return model.StrategyDetail{}, err
	}
	detail, err := strategyDetail(ctx, tx, id)
	if err != nil {
		return detail, err
	}
	return detail, mapError(tx.Commit(ctx))
}
func (r *StrategyRepository) Deploy(ctx context.Context, id, versionID, status string, expected int64, reason, actor, requestID string) (model.StrategyDetail, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	defer tx.Rollback(context.Background())
	s, err := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1 FOR UPDATE`, id))
	if err != nil {
		return model.StrategyDetail{}, err
	}
	if s.Version != expected {
		return model.StrategyDetail{}, apperror.ErrConflict
	}
	var selected *string
	if status == "ACTIVE" {
		var exists bool
		if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM veyra_strategy_version WHERE id=$1 AND strategy_id=$2)`, versionID, id).Scan(&exists); err != nil {
			return model.StrategyDetail{}, mapError(err)
		}
		if !exists {
			return model.StrategyDetail{}, apperror.ErrNotFound
		}
		selected = &versionID
	} else {
		selected = s.ActiveVersionID
	}
	_, err = tx.Exec(ctx, `UPDATE veyra_strategy SET status=$2,active_version_id=$3,version=version+1,updated_at=now() WHERE id=$1`, id, status, selected)
	if err != nil {
		return model.StrategyDetail{}, mapError(err)
	}
	if err := appendAudit(ctx, tx, actor, "STRATEGY_DEPLOYMENT_CHANGED", "strategy", id, requestID, map[string]any{"status": status, "version_id": selected, "reason": reason}); err != nil {
		return model.StrategyDetail{}, err
	}
	detail, err := strategyDetail(ctx, tx, id)
	if err != nil {
		return detail, err
	}
	return detail, mapError(tx.Commit(ctx))
}