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, ¶meters, &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)) }