package repository import ( "context" "encoding/json" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "git.sundynix.cn/Blizzard/Veyra/internal/apperror" "git.sundynix.cn/Blizzard/Veyra/internal/model" ) type ProviderRepository struct{ pool *pgxpool.Pool } func NewProviderRepository(pool *pgxpool.Pool) *ProviderRepository { return &ProviderRepository{pool: pool} } const providerColumns = `id,config_key,name,base_url,model_name,enabled,timeout_seconds,encrypted_api_key,version,last_test_status,last_tested_at,last_test_message,updated_by,created_at,updated_at,wire_api,model_catalog_url,disable_response_storage,requires_openai_auth,market_adapter,history_base_url,quote_base_url` func scanProvider(row pgx.Row) (model.ProviderConfig, error) { var c model.ProviderConfig err := row.Scan(&c.ID, &c.ConfigKey, &c.Name, &c.BaseURL, &c.ModelName, &c.Enabled, &c.TimeoutSeconds, &c.EncryptedAPIKey, &c.Version, &c.LastTestStatus, &c.LastTestedAt, &c.LastTestMessage, &c.UpdatedBy, &c.CreatedAt, &c.UpdatedAt, &c.WireAPI, &c.ModelCatalogURL, &c.DisableResponseStorage, &c.RequiresOpenAIAuth, &c.MarketAdapter, &c.HistoryBaseURL, &c.QuoteBaseURL) c.HasAPIKey = len(c.EncryptedAPIKey) > 0 return c, mapError(err) } func (r *ProviderRepository) Get(ctx context.Context, key string) (model.ProviderConfig, error) { return scanProvider(r.pool.QueryRow(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config WHERE config_key=$1`, key)) } func (r *ProviderRepository) List(ctx context.Context) ([]model.ProviderConfig, error) { rows, err := r.pool.Query(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config ORDER BY config_key`) if err != nil { return nil, mapError(err) } defer rows.Close() configs := make([]model.ProviderConfig, 0) for rows.Next() { c, err := scanProvider(rows) if err != nil { return nil, err } configs = append(configs, c) } return configs, mapError(rows.Err()) } func (r *ProviderRepository) Save(ctx context.Context, c model.ProviderConfig, expected int64, actor, requestID string, secretChanged bool) (model.ProviderConfig, error) { tx, err := r.pool.Begin(ctx) if err != nil { return model.ProviderConfig{}, mapError(err) } defer tx.Rollback(context.Background()) // A per-key lock also serializes two simultaneous first-time creates. if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, "veyra:provider:"+c.ConfigKey); err != nil { return model.ProviderConfig{}, mapError(err) } var version int64 err = tx.QueryRow(ctx, `SELECT version FROM veyra_provider_config WHERE config_key=$1`, c.ConfigKey).Scan(&version) if err != nil && err != pgx.ErrNoRows { return model.ProviderConfig{}, mapError(err) } if version != expected { return model.ProviderConfig{}, apperror.ErrConflict } c.Version = version + 1 if version == 0 { c.ID = uuid.NewString() _, err = tx.Exec(ctx, `INSERT INTO veyra_provider_config(id,config_key,name,base_url,model_name,enabled,timeout_seconds,encrypted_api_key,version,last_test_status,updated_by,wire_api,model_catalog_url,disable_response_storage,requires_openai_auth,market_adapter,history_base_url,quote_base_url) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18)`, c.ID, c.ConfigKey, c.Name, c.BaseURL, c.ModelName, c.Enabled, c.TimeoutSeconds, c.EncryptedAPIKey, c.Version, c.LastTestStatus, actor, c.WireAPI, c.ModelCatalogURL, c.DisableResponseStorage, c.RequiresOpenAIAuth, c.MarketAdapter, c.HistoryBaseURL, c.QuoteBaseURL) } else { _, err = tx.Exec(ctx, `UPDATE veyra_provider_config SET name=$2,base_url=$3,model_name=$4,enabled=$5,timeout_seconds=$6,encrypted_api_key=$7,version=$8,last_test_status=$9,last_tested_at=NULL,last_test_message='',updated_by=$10,wire_api=$11,model_catalog_url=$12,disable_response_storage=$13,requires_openai_auth=$14,market_adapter=$15,history_base_url=$16,quote_base_url=$17,updated_at=now() WHERE config_key=$1`, c.ConfigKey, c.Name, c.BaseURL, c.ModelName, c.Enabled, c.TimeoutSeconds, c.EncryptedAPIKey, c.Version, c.LastTestStatus, actor, c.WireAPI, c.ModelCatalogURL, c.DisableResponseStorage, c.RequiresOpenAIAuth, c.MarketAdapter, c.HistoryBaseURL, c.QuoteBaseURL) } if err != nil { return model.ProviderConfig{}, mapError(err) } c, err = scanProvider(tx.QueryRow(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config WHERE config_key=$1`, c.ConfigKey)) if err != nil { return c, err } snapshot, err := json.Marshal(c) if err != nil { return c, apperror.ErrUnavailable } if _, err := tx.Exec(ctx, `INSERT INTO veyra_provider_config_history(id,config_id,version,snapshot,encrypted_api_key,actor_id) VALUES($1,$2,$3,$4,$5,$6)`, uuid.NewString(), c.ID, c.Version, snapshot, c.EncryptedAPIKey, actor); err != nil { return c, mapError(err) } if err := appendAudit(ctx, tx, actor, "PROVIDER_CONFIG_SAVED", "provider_config", c.ID, requestID, map[string]any{"config_key": c.ConfigKey, "version": c.Version, "enabled": c.Enabled, "secret_changed": secretChanged}); err != nil { return c, err } if err := tx.Commit(ctx); err != nil { return c, mapError(err) } return c, nil } func (r *ProviderRepository) RecordTest(ctx context.Context, c model.ProviderConfig, status, message, actor, requestID string) error { tx, err := r.pool.Begin(ctx) if err != nil { return mapError(err) } defer tx.Rollback(context.Background()) tag, err := tx.Exec(ctx, `UPDATE veyra_provider_config SET last_test_status=$3,last_test_message=$4,last_tested_at=$5 WHERE id=$1 AND version=$2`, c.ID, c.Version, status, message, time.Now().UTC()) if err != nil { return mapError(err) } if tag.RowsAffected() != 1 { return apperror.ErrConflict } if err := appendAudit(ctx, tx, actor, "PROVIDER_TESTED", "provider_config", c.ID, requestID, map[string]any{"version": c.Version, "status": status}); err != nil { return err } return mapError(tx.Commit(ctx)) } func appendAudit(ctx context.Context, tx pgx.Tx, actor, action, objectType, objectID, requestID string, details any) error { encoded, err := json.Marshal(details) if err != nil { return apperror.ErrUnavailable } _, err = tx.Exec(ctx, `INSERT INTO veyra_audit_event(id,actor_id,action,object_type,object_id,request_id,details) VALUES($1,$2,$3,$4,$5,$6,$7)`, uuid.NewString(), actor, action, objectType, objectID, requestID, encoded) return mapError(err) } func (r *ProviderRepository) Audit(ctx context.Context) ([]model.AuditEvent, error) { rows, err := r.pool.Query(ctx, `SELECT id,actor_id,action,object_type,object_id,request_id,details,created_at FROM veyra_audit_event ORDER BY created_at DESC,id DESC LIMIT 100`) if err != nil { return nil, mapError(err) } defer rows.Close() list := make([]model.AuditEvent, 0) for rows.Next() { var a model.AuditEvent if err := rows.Scan(&a.ID, &a.ActorID, &a.Action, &a.ObjectType, &a.ObjectID, &a.RequestID, &a.Details, &a.CreatedAt); err != nil { return nil, mapError(err) } list = append(list, a) } return list, mapError(rows.Err()) }