259 lines
13 KiB
Go
259 lines
13 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
|
|
"time"
|
|
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/model"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type CandidateRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewCandidateRepository(pool *pgxpool.Pool) *CandidateRepository {
|
|
return &CandidateRepository{pool: pool}
|
|
}
|
|
|
|
// All successors of an automatically generated candidate retain the deployment binding.
|
|
const screeningEligibility = `NOT EXISTS(SELECT 1 FROM veyra_strategy_screening x JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id JOIN veyra_strategy_version sv ON sv.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=sv.strategy_id WHERE original.candidate_id=c.id AND (s.status<>'ACTIVE' OR s.active_version_id IS DISTINCT FROM sv.id))`
|
|
|
|
const candidateColumns = `r.id,r.candidate_id,r.revision,r.version,r.status,r.data_mode,r.snapshot,r.author_id,r.observed_at,r.valid_until,r.published_at,r.created_at,r.updated_at,(r.revision=c.latest_revision),COALESCE(c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now() AND ` + screeningEligibility + `,false),COALESCE((SELECT reason FROM veyra_candidate_review v WHERE v.revision_id=r.id AND v.decision='WITHDRAW' ORDER BY v.input_version DESC LIMIT 1),''),(SELECT jsonb_build_object('id',x.id,'strategy_version_id',x.strategy_version_id,'strategy_name',s.name,'strategy_revision',sv.revision,'symbol',x.symbol,'exchange',x.exchange,'bar_date',x.bar_date::text,'source_id',x.source_id,'source_version',x.source_version,'source_adapter',x.source_adapter,'observed_at',x.observed_at,'received_at',x.received_at,'input_hash',x.input_hash,'status',x.status,'candidate_revision_id',x.candidate_revision_id,'created_at',x.created_at) FROM veyra_strategy_screening x JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id JOIN veyra_strategy_version sv ON sv.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=sv.strategy_id WHERE original.candidate_id=c.id LIMIT 1),(SELECT jsonb_build_object('id',a.id,'candidate_revision_id',a.candidate_revision_id,'input_version',a.input_version,'policy_version',a.policy_version,'status',a.status,'primary_model',a.primary_model,'review_model',a.review_model,'primary_version',a.primary_version,'review_version',a.review_version,'analysis',a.analysis,'review',a.review,'message',a.message,'created_at',a.created_at,'updated_at',a.updated_at) FROM veyra_candidate_analysis a WHERE a.candidate_revision_id=r.id AND (a.input_version=r.version OR (r.status='PUBLISHED' AND a.input_version=r.version-1)) AND a.status IN ('APPROVED','PUBLISHED') ORDER BY a.updated_at DESC LIMIT 1)`
|
|
|
|
func scanCandidate(row pgx.Row) (model.CandidateRevision, error) {
|
|
var r model.CandidateRevision
|
|
var snapshot, screening, analysis []byte
|
|
err := row.Scan(&r.ID, &r.CandidateID, &r.Revision, &r.Version, &r.Status, &r.DataMode, &snapshot, &r.AuthorID, &r.ObservedAt, &r.ValidUntil, &r.PublishedAt, &r.CreatedAt, &r.UpdatedAt, &r.IsLatest, &r.IsActivePublication, &r.WithdrawalReason, &screening, &analysis)
|
|
if err != nil {
|
|
return r, mapError(err)
|
|
}
|
|
if err := json.Unmarshal(snapshot, &r.Snapshot); err != nil {
|
|
return r, apperror.ErrUnavailable
|
|
}
|
|
if len(screening) > 0 {
|
|
if err := json.Unmarshal(screening, &r.Screening); err != nil {
|
|
return r, apperror.ErrUnavailable
|
|
}
|
|
}
|
|
if len(analysis) > 0 {
|
|
if err := json.Unmarshal(analysis, &r.AIAnalysis); err != nil {
|
|
return r, apperror.ErrUnavailable
|
|
}
|
|
}
|
|
return r, nil
|
|
}
|
|
|
|
const candidateJoin = ` FROM veyra_candidate_revision r JOIN veyra_candidate c ON c.id=r.candidate_id `
|
|
|
|
func (r *CandidateRepository) List(ctx context.Context, admin bool, status string, limit int) ([]model.CandidateRevision, error) {
|
|
query := `SELECT ` + candidateColumns + candidateJoin + ` WHERE ($1 AND (r.revision=c.latest_revision OR c.published_revision_id=r.id) AND ($2='' OR r.status=$2)) OR (NOT $1 AND c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now() AND ` + screeningEligibility + `) ORDER BY r.created_at DESC,r.id DESC LIMIT $3`
|
|
rows, err := r.pool.Query(ctx, query, admin, status, limit)
|
|
if err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
list := make([]model.CandidateRevision, 0)
|
|
for rows.Next() {
|
|
v, err := scanCandidate(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, v)
|
|
}
|
|
return list, mapError(rows.Err())
|
|
}
|
|
func (r *CandidateRepository) Detail(ctx context.Context, id string, admin bool) (model.CandidateDetail, error) {
|
|
revision, err := scanCandidate(r.pool.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1 AND ($2 OR r.published_at IS NOT NULL)`, id, admin))
|
|
detail := model.CandidateDetail{CandidateRevision: revision, Reviews: make([]model.CandidateReview, 0)}
|
|
if err != nil {
|
|
return detail, err
|
|
}
|
|
// App readers receive the publication, not internal reviewer identities or deliberations.
|
|
if !admin {
|
|
return detail, nil
|
|
}
|
|
rows, err := r.pool.Query(ctx, `SELECT id,revision_id,reviewer_id,decision,input_version,reason,created_at FROM veyra_candidate_review WHERE revision_id=$1 ORDER BY input_version`, id)
|
|
if err != nil {
|
|
return detail, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var v model.CandidateReview
|
|
if err := rows.Scan(&v.ID, &v.RevisionID, &v.ReviewerID, &v.Decision, &v.InputVersion, &v.Reason, &v.CreatedAt); err != nil {
|
|
return detail, mapError(err)
|
|
}
|
|
detail.Reviews = append(detail.Reviews, v)
|
|
}
|
|
return detail, mapError(rows.Err())
|
|
}
|
|
|
|
// All mutations lock the candidate aggregate before its revisions, with the same lock order.
|
|
func lockCandidate(ctx context.Context, tx pgx.Tx, source string) (model.CandidateRevision, error) {
|
|
var parent string
|
|
if err := tx.QueryRow(ctx, `SELECT candidate_id FROM veyra_candidate_revision WHERE id=$1`, source).Scan(&parent); err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
// Match scanner lock order: deployment, then candidate aggregate.
|
|
rows, err := tx.Query(ctx, `SELECT s.id FROM veyra_strategy s JOIN veyra_strategy_version v ON v.strategy_id=s.id JOIN veyra_strategy_screening x ON x.strategy_version_id=v.id JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id WHERE original.candidate_id=$1 ORDER BY s.id FOR SHARE OF s`, parent)
|
|
if err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
for rows.Next() {
|
|
var id string
|
|
if err := rows.Scan(&id); err != nil {
|
|
rows.Close()
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
var locked string
|
|
if err := tx.QueryRow(ctx, `SELECT id FROM veyra_candidate WHERE id=$1 FOR UPDATE`, parent).Scan(&locked); err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
return scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1 FOR UPDATE OF r`, source))
|
|
}
|
|
func (r *CandidateRepository) Create(ctx context.Context, input model.CandidateDraftInput, actor, requestID, source string, expected int64) (model.CandidateRevision, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
created, err := createCandidate(ctx, tx, input, actor, requestID, source, expected)
|
|
if err != nil {
|
|
return created, err
|
|
}
|
|
return created, mapError(tx.Commit(ctx))
|
|
}
|
|
|
|
func createCandidate(ctx context.Context, tx pgx.Tx, input model.CandidateDraftInput, actor, requestID, source string, expected int64) (model.CandidateRevision, error) {
|
|
var err error
|
|
parent, revision := uuid.NewString(), 1
|
|
if source == "" {
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate(id,latest_revision) VALUES($1,1)`, parent)
|
|
} else {
|
|
previous, e := lockCandidate(ctx, tx, source)
|
|
if e != nil {
|
|
return model.CandidateRevision{}, e
|
|
}
|
|
if previous.Version != expected || !previous.IsLatest || previous.Status == "PENDING_REVIEW" {
|
|
return model.CandidateRevision{}, apperror.ErrConflict
|
|
}
|
|
parent, revision = previous.CandidateID, previous.Revision+1
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET latest_revision=$2 WHERE id=$1`, parent, revision)
|
|
}
|
|
if err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
id := uuid.NewString()
|
|
snapshot, err := json.Marshal(input.Content)
|
|
if err != nil {
|
|
return model.CandidateRevision{}, apperror.ErrInvalidInput
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate_revision(id,candidate_id,revision,status,data_mode,snapshot,author_id,observed_at,valid_until) VALUES($1,$2,$3,'DRAFT','simulation',$4,$5,$6,$7)`, id, parent, revision, snapshot, actor, input.ObservedAt, input.ValidUntil)
|
|
if err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, actor, "CANDIDATE_DRAFT_CREATED", "candidate_revision", id, requestID, map[string]any{"revision": revision, "source_revision_id": source}); err != nil {
|
|
return model.CandidateRevision{}, err
|
|
}
|
|
created, err := scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1`, id))
|
|
if err != nil {
|
|
return created, err
|
|
}
|
|
return created, nil
|
|
}
|
|
func appendCandidateEvent(ctx context.Context, tx pgx.Tx, r model.CandidateRevision, eventType string, details any) error {
|
|
payload, err := json.Marshal(details)
|
|
if err != nil {
|
|
return apperror.ErrUnavailable
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,'candidate_revision',$2,$3,$4,$5)`, uuid.NewString(), r.ID, r.Version, eventType, payload)
|
|
return mapError(err)
|
|
}
|
|
func (r *CandidateRepository) Transition(ctx context.Context, id, action string, input model.CandidateActionInput, actor, requestID string) (model.CandidateRevision, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return model.CandidateRevision{}, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
revision, err := transitionCandidate(ctx, tx, id, action, input, actor, requestID)
|
|
if err != nil {
|
|
return revision, err
|
|
}
|
|
return revision, mapError(tx.Commit(ctx))
|
|
}
|
|
|
|
func transitionCandidate(ctx context.Context, tx pgx.Tx, id, action string, input model.CandidateActionInput, actor, requestID string) (model.CandidateRevision, error) {
|
|
revision, err := lockCandidate(ctx, tx, id)
|
|
if err != nil {
|
|
return revision, err
|
|
}
|
|
if revision.Version != input.ExpectedVersion {
|
|
return revision, apperror.ErrConflict
|
|
}
|
|
status, err := model.NextStatus(revision, action, time.Now().UTC())
|
|
if err != nil {
|
|
return revision, err
|
|
}
|
|
if action == "PUBLISH" {
|
|
var eligible bool
|
|
if err := tx.QueryRow(ctx, `SELECT `+screeningEligibility+candidateJoin+` WHERE r.id=$1`, id).Scan(&eligible); err != nil {
|
|
return revision, mapError(err)
|
|
}
|
|
if !eligible {
|
|
return revision, apperror.ErrCandidateUnavailable
|
|
}
|
|
// Superseded publications keep their historical content, with an invalidation event.
|
|
previous, err := scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE c.id=$1 AND c.published_revision_id=r.id`, revision.CandidateID))
|
|
if err != nil && err != apperror.ErrNotFound {
|
|
return revision, err
|
|
}
|
|
if err == nil {
|
|
if err := appendCandidateEvent(ctx, tx, previous, "candidate.superseded", map[string]any{"replacement_revision_id": id}); err != nil {
|
|
return revision, err
|
|
}
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET published_revision_id=$2 WHERE id=$1`, revision.CandidateID, id)
|
|
if err != nil {
|
|
return revision, mapError(err)
|
|
}
|
|
} else if action == "WITHDRAW" {
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET published_revision_id=NULL WHERE id=$1`, revision.CandidateID)
|
|
if err != nil {
|
|
return revision, mapError(err)
|
|
}
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_candidate_revision SET status=$2::varchar(24),version=version+1,updated_at=now(),published_at=CASE WHEN $2::varchar(24)='PUBLISHED' THEN now() ELSE published_at END WHERE id=$1`, id, status)
|
|
if err != nil {
|
|
return revision, mapError(err)
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate_review(id,revision_id,reviewer_id,decision,input_version,reason) VALUES($1,$2,$3,$4,$5,$6)`, uuid.NewString(), id, actor, action, input.ExpectedVersion, input.Reason)
|
|
if err != nil {
|
|
return revision, mapError(err)
|
|
}
|
|
if err := appendAudit(ctx, tx, actor, "CANDIDATE_"+action, "candidate_revision", id, requestID, map[string]any{"reason": input.Reason, "input_version": input.ExpectedVersion, "status": status}); err != nil {
|
|
return revision, err
|
|
}
|
|
revision, err = scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1`, id))
|
|
if err != nil {
|
|
return revision, err
|
|
}
|
|
if action == "PUBLISH" || action == "WITHDRAW" {
|
|
eventType := "candidate.published"
|
|
if action == "WITHDRAW" {
|
|
eventType = "candidate.withdrawn"
|
|
}
|
|
if err := appendCandidateEvent(ctx, tx, revision, eventType, map[string]any{"revision_id": id, "version": revision.Version, "reason": input.Reason}); err != nil {
|
|
return revision, err
|
|
}
|
|
}
|
|
return revision, nil
|
|
}
|