Veyra/internal/repository/provider.go

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())
}