package store

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net"
	"strconv"
	"strings"
	"time"
	"unicode"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/publicerror"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"golang.org/x/crypto/bcrypt"
)

type Store struct {
	pool                   *pgxpool.Pool
	walletReservationLocks walletReservationLockSet
}

const (
	postgresApplicationName = "easyai-ai-gateway"
	postgresConnectTimeout  = 5 * time.Second
)

type PostgresPoolOptions struct {
	MaxConns                 int
	MinIdleConns             int
	MaxConnIdleTime          time.Duration
	IdleInTransactionTimeout time.Duration
	LockTimeout              time.Duration
}

func defaultAPIKeyScopes() []string {
	return []string{"chat", "embedding", "rerank", "image", "image_vectorize", "video", "video_enhance", "music", "audio", "voice_clone"}
}

func normalizeAPIKeyScopes(scopes []string) []string {
	out := make([]string, 0, len(scopes))
	seen := make(map[string]struct{}, len(scopes))
	for _, scope := range scopes {
		scope = strings.ToLower(strings.TrimSpace(scope))
		if scope == "" {
			continue
		}
		if scope == "*" {
			scope = "all"
		}
		if _, ok := seen[scope]; ok {
			continue
		}
		seen[scope] = struct{}{}
		out = append(out, scope)
	}
	return out
}

var (
	ErrInvalidCredentials            = errors.New("invalid account or password")
	ErrInvalidInvitation             = errors.New("invalid or expired invitation code")
	ErrInvalidAPIKeyScopes           = errors.New("api key scopes must not be empty")
	ErrInvalidIdentityBatch          = errors.New("identity batch action or ids are invalid")
	ErrIdentityBatchTargetNotFound   = errors.New("one or more identity batch targets were not found")
	ErrAccessRuleResourceDenied      = errors.New("access rule resource is not available")
	ErrInsufficientWalletBalance     = errors.New("insufficient wallet balance")
	ErrLocalUserRequired             = errors.New("local gateway user is required")
	ErrWalletBalanceUnchanged        = errors.New("wallet balance unchanged")
	ErrBalanceBelowFrozen            = errors.New("wallet balance cannot be below frozen balance")
	ErrInvalidWalletAmount           = errors.New("wallet amount must be a decimal with at most nine fractional digits")
	ErrIdempotencyKeyReused          = errors.New("idempotency key was reused for a different request")
	ErrExternalTaskIDReused          = errors.New("external task id was reused")
	ErrTaskExecutionLeaseUnavailable = errors.New("task execution lease is unavailable")
	ErrTaskExecutionLeaseLost        = errors.New("task execution lease was lost")
	ErrTaskExecutionFinished         = errors.New("task execution already finished")
	ErrTaskExecutionManualReview     = errors.New("task execution requires manual review")
	ErrProtectedDefault              = errors.New("protected default resource cannot be deleted")
	ErrBaseModelInUse                = errors.New("base model is still referenced by platform models")
	ErrModelAliasConflict            = errors.New("model compatibility alias conflicts with another model identity")
	ErrUserAlreadyExists             = errors.New("user already exists")
	ErrWeakPassword                  = errors.New("password must be at least 8 characters")
)

func Connect(ctx context.Context, databaseURL string) (*Store, error) {
	return ConnectWithMaxConns(ctx, databaseURL, 0)
}

func ConnectWithMaxConns(ctx context.Context, databaseURL string, maxConns int) (*Store, error) {
	return ConnectWithPoolOptions(ctx, databaseURL, PostgresPoolOptions{MaxConns: maxConns})
}

func ConnectWithPoolOptions(ctx context.Context, databaseURL string, options PostgresPoolOptions) (*Store, error) {
	config, err := postgresPoolConfigWithOptions(databaseURL, options)
	if err != nil {
		return nil, err
	}
	pool, err := pgxpool.NewWithConfig(ctx, config)
	if err != nil {
		return nil, err
	}
	if err := warmPostgresPool(ctx, pool, int(config.MinIdleConns)); err != nil {
		pool.Close()
		return nil, err
	}
	return &Store{pool: pool}, nil
}

func postgresPoolConfig(databaseURL string, maxConns ...int) (*pgxpool.Config, error) {
	options := PostgresPoolOptions{}
	if len(maxConns) > 0 {
		options.MaxConns = maxConns[0]
	}
	return postgresPoolConfigWithOptions(databaseURL, options)
}

func postgresPoolConfigWithOptions(databaseURL string, options PostgresPoolOptions) (*pgxpool.Config, error) {
	config, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		return nil, err
	}
	config.ConnConfig.ConnectTimeout = postgresConnectTimeout
	config.ConnConfig.RuntimeParams["application_name"] = postgresApplicationName
	if options.IdleInTransactionTimeout > 0 {
		config.ConnConfig.RuntimeParams["idle_in_transaction_session_timeout"] = postgresDuration(options.IdleInTransactionTimeout)
	}
	if options.LockTimeout > 0 {
		config.ConnConfig.RuntimeParams["lock_timeout"] = postgresDuration(options.LockTimeout)
	}
	if options.MaxConns > 0 {
		config.MaxConns = int32(options.MaxConns)
	}
	config.MinIdleConns = int32(min(max(options.MinIdleConns, 0), int(config.MaxConns)))
	if options.MaxConnIdleTime > 0 {
		config.MaxConnIdleTime = options.MaxConnIdleTime
	}
	return config, nil
}

func postgresDuration(value time.Duration) string {
	return strconv.FormatInt(max(value.Milliseconds(), 1), 10) + "ms"
}

func warmPostgresPool(ctx context.Context, pool *pgxpool.Pool, count int) error {
	if count <= 0 {
		return pool.Ping(ctx)
	}
	connections := make([]*pgxpool.Conn, 0, count)
	defer func() {
		for _, connection := range connections {
			connection.Release()
		}
	}()
	for len(connections) < count {
		connection, err := pool.Acquire(ctx)
		if err != nil {
			return err
		}
		connections = append(connections, connection)
	}
	return nil
}

func IsPostgresUnavailable(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	var connectError *pgconn.ConnectError
	if errors.As(err, &connectError) {
		return true
	}
	var networkError net.Error
	if errors.As(err, &networkError) {
		return true
	}
	var postgresError *pgconn.PgError
	if errors.As(err, &postgresError) {
		return strings.HasPrefix(postgresError.Code, "08") || postgresError.Code == "53300" || strings.HasPrefix(postgresError.Code, "57P0")
	}
	return pgconn.SafeToRetry(err)
}

func IsPostgresLockTimeout(err error) bool {
	var postgresError *pgconn.PgError
	return errors.As(err, &postgresError) && postgresError.Code == "55P03"
}

func (s *Store) Close() {
	s.pool.Close()
}

func (s *Store) Ping(ctx context.Context) error {
	return s.pool.Ping(ctx)
}

func (s *Store) Pool() *pgxpool.Pool {
	return s.pool
}

type Platform struct {
	ID                    string         `json:"id"`
	Provider              string         `json:"provider"`
	PlatformKey           string         `json:"platformKey"`
	Name                  string         `json:"name"`
	InternalName          string         `json:"internalName,omitempty"`
	BaseURL               string         `json:"baseUrl,omitempty"`
	AuthType              string         `json:"authType"`
	Status                string         `json:"status"`
	Priority              int            `json:"priority"`
	DynamicPriority       *int           `json:"dynamicPriority,omitempty"`
	EffectivePriority     int            `json:"effectivePriority"`
	DefaultPricingMode    string         `json:"defaultPricingMode"`
	DefaultDiscountFactor float64        `json:"defaultDiscountFactor"`
	PricingRuleSetID      string         `json:"pricingRuleSetId,omitempty"`
	Config                map[string]any `json:"config,omitempty"`
	CredentialsPreview    map[string]any `json:"credentialsPreview,omitempty"`
	RetryPolicy           map[string]any `json:"retryPolicy,omitempty"`
	RateLimitPolicy       map[string]any `json:"rateLimitPolicy,omitempty"`
	CooldownUntil         string         `json:"cooldownUntil,omitempty"`
	CreatedAt             time.Time      `json:"createdAt"`
	UpdatedAt             time.Time      `json:"updatedAt"`
}

type CreatePlatformInput struct {
	Provider              string         `json:"provider"`
	PlatformKey           string         `json:"platformKey"`
	Name                  string         `json:"name"`
	InternalName          string         `json:"internalName"`
	BaseURL               string         `json:"baseUrl"`
	AuthType              string         `json:"authType"`
	Credentials           map[string]any `json:"credentials"`
	Config                map[string]any `json:"config"`
	RetryPolicy           map[string]any `json:"retryPolicy"`
	RateLimitPolicy       map[string]any `json:"rateLimitPolicy"`
	DefaultPricingMode    string         `json:"defaultPricingMode"`
	DefaultDiscountFactor float64        `json:"defaultDiscountFactor"`
	PricingRuleSetID      string         `json:"pricingRuleSetId"`
	Priority              int            `json:"priority"`
	Status                string         `json:"status"`
}

type CreateAPIKeyInput struct {
	Name      string   `json:"name"`
	Scopes    []string `json:"scopes"`
	ExpiresAt string   `json:"expiresAt"`
}

type UpdateAPIKeyScopesInput struct {
	Scopes []string `json:"scopes"`
}

type APIKey struct {
	ID              string         `json:"id"`
	GatewayTenantID string         `json:"gatewayTenantId,omitempty"`
	GatewayUserID   string         `json:"gatewayUserId"`
	TenantID        string         `json:"tenantId,omitempty"`
	TenantKey       string         `json:"tenantKey,omitempty"`
	UserID          string         `json:"userId,omitempty"`
	KeyPrefix       string         `json:"keyPrefix"`
	Secret          string         `json:"secret,omitempty"`
	Name            string         `json:"name"`
	Scopes          []string       `json:"scopes,omitempty"`
	UserGroupID     string         `json:"userGroupId,omitempty"`
	RateLimitPolicy map[string]any `json:"rateLimitPolicy,omitempty"`
	QuotaPolicy     map[string]any `json:"quotaPolicy,omitempty"`
	Status          string         `json:"status"`
	ExpiresAt       string         `json:"expiresAt,omitempty"`
	LastUsedAt      string         `json:"lastUsedAt,omitempty"`
	CreatedAt       time.Time      `json:"createdAt"`
	UpdatedAt       time.Time      `json:"updatedAt"`
}

type PlayableAPIKey struct {
	APIKey
	Secret string `json:"secret"`
}

type CreatedAPIKey struct {
	APIKey APIKey `json:"apiKey"`
	Secret string `json:"secret"`
}

type PlatformModel struct {
	ID                string `json:"id"`
	PlatformID        string `json:"platformId"`
	BaseModelID       string `json:"baseModelId,omitempty"`
	Provider          string `json:"provider,omitempty"`
	PlatformName      string `json:"platformName,omitempty"`
	ModelName         string `json:"modelName"`
	ProviderModelName string `json:"providerModelName,omitempty"`
	// Deprecated: use ModelName. This field mirrors the canonical invocation name during the compatibility window.
	ModelAlias               string         `json:"modelAlias,omitempty"`
	LegacyAliases            StringList     `json:"legacyAliases,omitempty"`
	ModelType                StringList     `json:"modelType"`
	DisplayName              string         `json:"displayName"`
	CapabilityOverride       map[string]any `json:"capabilityOverride,omitempty"`
	Capabilities             map[string]any `json:"capabilities,omitempty"`
	BaseCapabilities         map[string]any `json:"-"`
	BaseBillingConfig        map[string]any `json:"-"`
	BaseRateLimitPolicy      map[string]any `json:"-"`
	BaseRuntimePolicySetID   string         `json:"-"`
	BasePricingRuleSetID     string         `json:"-"`
	PlatformPricingRuleSetID string         `json:"-"`
	PricingMode              string         `json:"pricingMode"`
	DiscountFactor           float64        `json:"discountFactor,omitempty"`
	PricingRuleSetID         string         `json:"pricingRuleSetId,omitempty"`
	BillingConfigOverride    map[string]any `json:"billingConfigOverride,omitempty"`
	BillingConfig            map[string]any `json:"billingConfig,omitempty"`
	PermissionConfig         map[string]any `json:"permissionConfig,omitempty"`
	RetryPolicy              map[string]any `json:"retryPolicy,omitempty"`
	RateLimitPolicy          map[string]any `json:"rateLimitPolicy,omitempty"`
	RateLimitPolicyMode      string         `json:"rateLimitPolicyMode" enums:"inherit,override"`
	RuntimePolicySetID       string         `json:"runtimePolicySetId,omitempty"`
	RuntimePolicyOverride    map[string]any `json:"runtimePolicyOverride,omitempty"`
	CooldownUntil            string         `json:"cooldownUntil,omitempty"`
	Enabled                  bool           `json:"enabled"`
	CreatedAt                time.Time      `json:"createdAt"`
	UpdatedAt                time.Time      `json:"updatedAt"`
}

type AccessRule struct {
	ID                 string         `json:"id"`
	SubjectType        string         `json:"subjectType"`
	SubjectID          string         `json:"subjectId"`
	ResourceType       string         `json:"resourceType"`
	ResourceID         string         `json:"resourceId"`
	Effect             string         `json:"effect"`
	Priority           int            `json:"priority"`
	MinPermissionLevel int            `json:"minPermissionLevel"`
	Conditions         map[string]any `json:"conditions,omitempty"`
	Metadata           map[string]any `json:"metadata,omitempty"`
	Status             string         `json:"status"`
	CreatedAt          time.Time      `json:"createdAt"`
	UpdatedAt          time.Time      `json:"updatedAt"`
}

type CatalogProvider struct {
	ID                     string         `json:"id"`
	ProviderKey            string         `json:"providerKey"`
	Code                   string         `json:"code"`
	DisplayName            string         `json:"displayName"`
	ProviderType           string         `json:"providerType"`
	IconPath               string         `json:"iconPath,omitempty"`
	DefaultBaseURL         string         `json:"defaultBaseUrl,omitempty"`
	DefaultAuthType        string         `json:"defaultAuthType,omitempty"`
	Source                 string         `json:"source,omitempty"`
	CapabilitySchema       map[string]any `json:"capabilitySchema,omitempty"`
	DefaultRateLimitPolicy map[string]any `json:"defaultRateLimitPolicy,omitempty"`
	Metadata               map[string]any `json:"metadata,omitempty"`
	Status                 string         `json:"status"`
	CreatedAt              time.Time      `json:"createdAt"`
	UpdatedAt              time.Time      `json:"updatedAt"`
}

type BaseModel struct {
	ID                string     `json:"id"`
	ProviderKey       string     `json:"providerKey"`
	CanonicalModelKey string     `json:"canonicalModelKey"`
	InvocationName    string     `json:"invocationName"`
	ProviderModelName string     `json:"providerModelName"`
	ModelType         StringList `json:"modelType"`
	// Deprecated: use InvocationName.
	ModelAlias             string         `json:"modelAlias"`
	DisplayName            string         `json:"displayName"`
	LegacyAliases          StringList     `json:"legacyAliases,omitempty"`
	ReferenceCount         int            `json:"referenceCount"`
	Capabilities           map[string]any `json:"capabilities,omitempty"`
	BaseBillingConfig      map[string]any `json:"baseBillingConfig,omitempty"`
	DefaultRateLimitPolicy map[string]any `json:"defaultRateLimitPolicy,omitempty"`
	PricingRuleSetID       string         `json:"pricingRuleSetId,omitempty"`
	RuntimePolicySetID     string         `json:"runtimePolicySetId,omitempty"`
	RuntimePolicyOverride  map[string]any `json:"runtimePolicyOverride,omitempty"`
	Metadata               map[string]any `json:"metadata,omitempty"`
	CatalogType            string         `json:"catalogType"`
	DefaultSnapshot        map[string]any `json:"defaultSnapshot,omitempty"`
	CustomizedAt           string         `json:"customizedAt,omitempty"`
	PricingVersion         int            `json:"pricingVersion"`
	Status                 string         `json:"status"`
	CreatedAt              time.Time      `json:"createdAt"`
	UpdatedAt              time.Time      `json:"updatedAt"`
}

type RuntimePolicySet struct {
	ID                string         `json:"id"`
	PolicyKey         string         `json:"policyKey"`
	Name              string         `json:"name"`
	Description       string         `json:"description,omitempty"`
	RateLimitPolicy   map[string]any `json:"rateLimitPolicy,omitempty"`
	RetryPolicy       map[string]any `json:"retryPolicy,omitempty"`
	AutoDisablePolicy map[string]any `json:"autoDisablePolicy,omitempty"`
	DegradePolicy     map[string]any `json:"degradePolicy,omitempty"`
	Metadata          map[string]any `json:"metadata,omitempty"`
	Status            string         `json:"status"`
	CreatedAt         time.Time      `json:"createdAt"`
	UpdatedAt         time.Time      `json:"updatedAt"`
}

type RunnerPolicy struct {
	ID                   string         `json:"id"`
	PolicyKey            string         `json:"policyKey"`
	Name                 string         `json:"name"`
	Description          string         `json:"description,omitempty"`
	FailoverPolicy       map[string]any `json:"failoverPolicy,omitempty"`
	HardStopPolicy       map[string]any `json:"hardStopPolicy,omitempty"`
	PriorityDemotePolicy map[string]any `json:"priorityDemotePolicy,omitempty"`
	SingleSourcePolicy   map[string]any `json:"singleSourcePolicy,omitempty"`
	CacheAffinityPolicy  map[string]any `json:"cacheAffinityPolicy,omitempty"`
	Metadata             map[string]any `json:"metadata,omitempty"`
	Status               string         `json:"status"`
	CreatedAt            time.Time      `json:"createdAt"`
	UpdatedAt            time.Time      `json:"updatedAt"`
}

type PricingRule struct {
	ID              string         `json:"id"`
	RuleSetID       string         `json:"ruleSetId,omitempty"`
	RuleKey         string         `json:"ruleKey"`
	DisplayName     string         `json:"displayName"`
	ScopeType       string         `json:"scopeType"`
	ScopeID         string         `json:"scopeId,omitempty"`
	ResourceType    string         `json:"resourceType"`
	Unit            string         `json:"unit"`
	BasePrice       float64        `json:"basePrice"`
	IsFree          bool           `json:"isFree"`
	Currency        string         `json:"currency"`
	BaseWeight      map[string]any `json:"baseWeight,omitempty"`
	DynamicWeight   map[string]any `json:"dynamicWeight,omitempty"`
	CalculatorType  string         `json:"calculatorType"`
	DimensionSchema map[string]any `json:"dimensionSchema,omitempty"`
	FormulaConfig   map[string]any `json:"formulaConfig,omitempty"`
	Priority        int            `json:"priority"`
	Status          string         `json:"status"`
	Metadata        map[string]any `json:"metadata,omitempty"`
	EffectiveFrom   string         `json:"effectiveFrom,omitempty"`
	EffectiveTo     string         `json:"effectiveTo,omitempty"`
	CreatedAt       time.Time      `json:"createdAt"`
	UpdatedAt       time.Time      `json:"updatedAt"`
}

type PricingRuleSet struct {
	ID          string         `json:"id"`
	RuleSetKey  string         `json:"ruleSetKey"`
	Name        string         `json:"name"`
	Description string         `json:"description,omitempty"`
	Category    string         `json:"category"`
	Currency    string         `json:"currency"`
	Status      string         `json:"status"`
	Metadata    map[string]any `json:"metadata,omitempty"`
	Rules       []PricingRule  `json:"rules,omitempty"`
	CreatedAt   time.Time      `json:"createdAt"`
	UpdatedAt   time.Time      `json:"updatedAt"`
}

type GatewayTenant struct {
	ID                 string         `json:"id"`
	TenantKey          string         `json:"tenantKey"`
	Source             string         `json:"source"`
	ExternalTenantID   string         `json:"externalTenantId,omitempty"`
	Name               string         `json:"name"`
	Description        string         `json:"description,omitempty"`
	DefaultUserGroupID string         `json:"defaultUserGroupId,omitempty"`
	PlanKey            string         `json:"planKey,omitempty"`
	BillingProfile     map[string]any `json:"billingProfile,omitempty"`
	RateLimitPolicy    map[string]any `json:"rateLimitPolicy,omitempty"`
	AuthPolicy         map[string]any `json:"authPolicy,omitempty"`
	Metadata           map[string]any `json:"metadata,omitempty"`
	Status             string         `json:"status"`
	SyncedAt           string         `json:"syncedAt,omitempty"`
	SourceUpdatedAt    string         `json:"sourceUpdatedAt,omitempty"`
	CreatedAt          time.Time      `json:"createdAt"`
	UpdatedAt          time.Time      `json:"updatedAt"`
}

type LocalRegisterInput struct {
	Username       string `json:"username"`
	Email          string `json:"email"`
	Password       string `json:"password"`
	DisplayName    string `json:"displayName"`
	InvitationCode string `json:"invitationCode"`
}

type LocalLoginInput struct {
	Account  string `json:"account"`
	Password string `json:"password"`
}

type GatewayUser struct {
	ID                 string                 `json:"id"`
	UserKey            string                 `json:"userKey"`
	Source             string                 `json:"source"`
	ExternalUserID     string                 `json:"externalUserId,omitempty"`
	Username           string                 `json:"username"`
	DisplayName        string                 `json:"displayName,omitempty"`
	Email              string                 `json:"email,omitempty"`
	Phone              string                 `json:"phone,omitempty"`
	AvatarURL          string                 `json:"avatarUrl,omitempty"`
	GatewayTenantID    string                 `json:"gatewayTenantId,omitempty"`
	TenantID           string                 `json:"tenantId,omitempty"`
	TenantKey          string                 `json:"tenantKey,omitempty"`
	DefaultUserGroupID string                 `json:"defaultUserGroupId,omitempty"`
	Roles              []string               `json:"roles,omitempty"`
	AuthProfile        map[string]any         `json:"authProfile,omitempty"`
	Metadata           map[string]any         `json:"metadata,omitempty"`
	WalletAccounts     []GatewayWalletAccount `json:"walletAccounts,omitempty"`
	Status             string                 `json:"status"`
	LastLoginAt        string                 `json:"lastLoginAt,omitempty"`
	SyncedAt           string                 `json:"syncedAt,omitempty"`
	SourceUpdatedAt    string                 `json:"sourceUpdatedAt,omitempty"`
	CreatedAt          time.Time              `json:"createdAt"`
	UpdatedAt          time.Time              `json:"updatedAt"`
}

type UserGroup struct {
	ID                     string         `json:"id"`
	GroupKey               string         `json:"groupKey"`
	Name                   string         `json:"name"`
	Description            string         `json:"description,omitempty"`
	Source                 string         `json:"source"`
	Priority               int            `json:"priority"`
	RechargeDiscountPolicy map[string]any `json:"rechargeDiscountPolicy,omitempty"`
	BillingDiscountPolicy  map[string]any `json:"billingDiscountPolicy,omitempty"`
	RateLimitPolicy        map[string]any `json:"rateLimitPolicy,omitempty"`
	QuotaPolicy            map[string]any `json:"quotaPolicy,omitempty"`
	Metadata               map[string]any `json:"metadata,omitempty"`
	Status                 string         `json:"status"`
	CreatedAt              time.Time      `json:"createdAt"`
	UpdatedAt              time.Time      `json:"updatedAt"`
}

type RateLimitWindow struct {
	ScopeType     string    `json:"scopeType"`
	ScopeKey      string    `json:"scopeKey"`
	Metric        string    `json:"metric"`
	WindowStart   time.Time `json:"windowStart"`
	LimitValue    float64   `json:"limitValue"`
	UsedValue     float64   `json:"usedValue"`
	ReservedValue float64   `json:"reservedValue"`
	ResetAt       time.Time `json:"resetAt"`
	UpdatedAt     time.Time `json:"updatedAt"`
}

type CreateTaskInput struct {
	Kind                   string                `json:"kind"`
	Model                  string                `json:"model"`
	ExternalTaskID         string                `json:"externalTaskId,omitempty"`
	RunMode                string                `json:"runMode"`
	AcceptanceRunID        string                `json:"acceptanceRunId,omitempty"`
	Async                  bool                  `json:"async"`
	Request                map[string]any        `json:"request"`
	ConversationID         string                `json:"conversationId"`
	NewMessageCount        int                   `json:"newMessageCount"`
	MessageRefs            []TaskMessageRefInput `json:"messageRefs"`
	IdempotencyKeyHash     string                `json:"-"`
	IdempotencyRequestHash string                `json:"-"`
}

type CreateTaskResult struct {
	Task     GatewayTask
	Replayed bool
}

type GatewayTask struct {
	ID                            string             `json:"id"`
	ExternalTaskID                string             `json:"externalTaskId,omitempty"`
	Kind                          string             `json:"kind"`
	RunMode                       string             `json:"runMode"`
	AcceptanceRunID               string             `json:"acceptanceRunId,omitempty"`
	UserID                        string             `json:"userId"`
	GatewayUserID                 string             `json:"gatewayUserId,omitempty"`
	UserSource                    string             `json:"userSource,omitempty"`
	GatewayTenantID               string             `json:"gatewayTenantId,omitempty"`
	TenantID                      string             `json:"tenantId,omitempty"`
	TenantKey                     string             `json:"tenantKey,omitempty"`
	APIKeyID                      string             `json:"apiKeyId,omitempty"`
	APIKeyName                    string             `json:"apiKeyName,omitempty"`
	APIKeyPrefix                  string             `json:"apiKeyPrefix,omitempty"`
	UserGroupID                   string             `json:"userGroupId,omitempty"`
	UserGroupKey                  string             `json:"userGroupKey,omitempty"`
	Model                         string             `json:"model"`
	ModelType                     string             `json:"modelType,omitempty"`
	RequestedModel                string             `json:"requestedModel,omitempty"`
	ResolvedModel                 string             `json:"resolvedModel,omitempty"`
	RequestID                     string             `json:"requestId,omitempty"`
	ConversationID                string             `json:"conversationId,omitempty"`
	NewMessageCount               int                `json:"newMessageCount,omitempty"`
	Request                       map[string]any     `json:"request,omitempty"`
	AsyncMode                     bool               `json:"asyncMode"`
	RiverJobID                    int64              `json:"riverJobId,omitempty"`
	Status                        string             `json:"status"`
	QueueKey                      string             `json:"-"`
	Priority                      int                `json:"-"`
	Cancellable                   *bool              `json:"cancellable,omitempty"`
	Submitted                     *bool              `json:"submitted,omitempty"`
	Message                       string             `json:"message,omitempty"`
	AttemptCount                  int                `json:"attemptCount"`
	RemoteTaskID                  string             `json:"remoteTaskId,omitempty"`
	RemoteTaskPayload             map[string]any     `json:"-"`
	Result                        map[string]any     `json:"result,omitempty"`
	Billings                      []any              `json:"billings,omitempty"`
	Usage                         map[string]any     `json:"usage"`
	Metrics                       map[string]any     `json:"metrics"`
	BillingSummary                map[string]any     `json:"billingSummary"`
	FinalChargeAmount             float64            `json:"finalChargeAmount"`
	BillingVersion                string             `json:"billingVersion"`
	BillingStatus                 string             `json:"billingStatus"`
	BillingCurrency               string             `json:"billingCurrency"`
	PricingSnapshot               map[string]any     `json:"pricingSnapshot,omitempty"`
	RequestFingerprint            string             `json:"requestFingerprint,omitempty"`
	ReservationAmount             float64            `json:"reservationAmount"`
	ExecutionToken                string             `json:"-"`
	ExecutionLeaseUntil           string             `json:"executionLeaseExpiresAt,omitempty"`
	BillingUpdatedAt              string             `json:"billingUpdatedAt,omitempty"`
	BillingSettledAt              string             `json:"billingSettledAt,omitempty"`
	ResponseStartedAt             string             `json:"responseStartedAt,omitempty"`
	ResponseFinishedAt            string             `json:"responseFinishedAt,omitempty"`
	ResponseDurationMS            int64              `json:"responseDurationMs"`
	FinishedAt                    string             `json:"finishedAt,omitempty"`
	Error                         string             `json:"error,omitempty"`
	ErrorCode                     string             `json:"errorCode,omitempty"`
	ErrorMessage                  string             `json:"errorMessage,omitempty"`
	PublicError                   *publicerror.Error `json:"publicError,omitempty"`
	CompatibilityProtocol         string             `json:"compatibilityProtocol,omitempty"`
	CompatibilityPublicID         string             `json:"compatibilityPublicId,omitempty"`
	CompatibilitySourceProtocol   string             `json:"compatibilitySourceProtocol,omitempty"`
	CompatibilitySubmitHTTPStatus int                `json:"compatibilitySubmitHttpStatus,omitempty"`
	CompatibilitySubmitHeaders    map[string]any     `json:"compatibilitySubmitHeaders,omitempty"`
	CompatibilitySubmitBody       map[string]any     `json:"compatibilitySubmitBody,omitempty"`
	AssignedPoolID                string             `json:"-"`
	AssignedWorkerID              string             `json:"-"`
	RouteProfileKey               string             `json:"-"`
	RoutingPlatformID             string             `json:"-"`
	RoutingPlatformModelID        string             `json:"-"`
	RoutingVersion                string             `json:"-"`
	RoutingReason                 string             `json:"-"`
	RoutingSnapshot               map[string]any     `json:"-"`
	SubmissionState               string             `json:"-"`
	Attempts                      []TaskAttempt      `json:"attempts,omitempty"`
	CreatedAt                     time.Time          `json:"createdAt"`
	UpdatedAt                     time.Time          `json:"updatedAt"`
}

const gatewayTaskColumns = `
id::text, COALESCE(external_task_id, ''), kind, run_mode, COALESCE(acceptance_run_id::text, ''),
user_id, COALESCE(gateway_user_id::text, ''), user_source,
COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''),
COALESCE(api_key_id, ''), COALESCE(api_key_name, ''), COALESCE(api_key_prefix, ''),
COALESCE(user_group_id::text, ''), COALESCE(user_group_key, ''), model,
COALESCE(model_type, ''), COALESCE(requested_model, ''), COALESCE(resolved_model, ''), COALESCE(request_id, ''),
COALESCE(conversation_id::text, ''), COALESCE(new_message_count, 0),
request, COALESCE(async_mode, false), COALESCE(river_job_id, 0), status,
COALESCE(queue_key, 'default'), COALESCE(priority, 100), COALESCE(attempt_count, 0),
COALESCE(remote_task_id, ''), COALESCE(remote_task_payload, '{}'::jsonb),
COALESCE(result, '{}'::jsonb), COALESCE(billings, '[]'::jsonb),
COALESCE(usage, '{}'::jsonb), COALESCE(metrics, '{}'::jsonb), COALESCE(billing_summary, '{}'::jsonb),
COALESCE(final_charge_amount, 0)::float8, billing_version, billing_status, billing_currency,
COALESCE(pricing_snapshot, '{}'::jsonb), COALESCE(request_fingerprint, ''),
COALESCE(reservation_amount, 0)::float8, COALESCE(execution_token::text, ''),
COALESCE(execution_lease_expires_at::text, ''), COALESCE(billing_updated_at::text, ''),
COALESCE(billing_settled_at::text, ''), COALESCE(response_started_at::text, ''),
COALESCE(response_finished_at::text, ''), COALESCE(response_duration_ms, 0), COALESCE(error, ''),
COALESCE(error_code, ''), COALESCE(error_message, ''), COALESCE(public_error, '{}'::jsonb),
COALESCE(compatibility_protocol, ''), COALESCE(compatibility_public_id, ''),
COALESCE(compatibility_source_protocol, ''), COALESCE(compatibility_submit_http_status, 0),
COALESCE(compatibility_submit_headers, '{}'::jsonb), COALESCE(compatibility_submit_body, '{}'::jsonb),
COALESCE(assigned_pool_id, ''), COALESCE(assigned_worker_id, ''), COALESCE(route_profile_key, ''),
COALESCE(routing_platform_id::text, ''), COALESCE(routing_platform_model_id::text, ''),
COALESCE(routing_version, ''), COALESCE(routing_reason, ''), COALESCE(routing_snapshot, '{}'::jsonb),
COALESCE(submission_state, 'not_started'),
created_at, updated_at, COALESCE(finished_at::text, '')`

type TaskEvent struct {
	ID            string         `json:"id"`
	TaskID        string         `json:"taskId"`
	PlatformID    string         `json:"-"`
	SkippedReason string         `json:"-"`
	Seq           int64          `json:"seq"`
	EventType     string         `json:"eventType"`
	Status        string         `json:"status,omitempty"`
	Phase         string         `json:"phase,omitempty"`
	Progress      float64        `json:"progress,omitempty"`
	Message       string         `json:"message,omitempty"`
	Payload       map[string]any `json:"payload,omitempty"`
	Simulated     bool           `json:"simulated"`
	CreatedAt     time.Time      `json:"createdAt"`
}

type TaskAttempt struct {
	ID                          string             `json:"id"`
	TaskID                      string             `json:"taskId"`
	AttemptNo                   int                `json:"attemptNo"`
	PlatformID                  string             `json:"platformId,omitempty"`
	PlatformName                string             `json:"platformName,omitempty"`
	Provider                    string             `json:"provider,omitempty"`
	PlatformModelID             string             `json:"platformModelId,omitempty"`
	ModelName                   string             `json:"modelName,omitempty"`
	ProviderModelName           string             `json:"providerModelName,omitempty"`
	ModelAlias                  string             `json:"modelAlias,omitempty"`
	ModelType                   string             `json:"modelType,omitempty"`
	ClientID                    string             `json:"clientId,omitempty"`
	QueueKey                    string             `json:"queueKey"`
	Status                      string             `json:"status"`
	Retryable                   bool               `json:"retryable"`
	Simulated                   bool               `json:"simulated"`
	RequestID                   string             `json:"requestId,omitempty"`
	StatusCode                  int                `json:"statusCode,omitempty"`
	Usage                       map[string]any     `json:"usage,omitempty"`
	Metrics                     map[string]any     `json:"metrics,omitempty"`
	RequestSnapshot             map[string]any     `json:"requestSnapshot,omitempty"`
	ResponseSnapshot            map[string]any     `json:"responseSnapshot,omitempty"`
	ResponseStartedAt           string             `json:"responseStartedAt,omitempty"`
	ResponseFinishedAt          string             `json:"responseFinishedAt,omitempty"`
	ResponseDurationMS          int64              `json:"responseDurationMs"`
	PricingSnapshot             map[string]any     `json:"pricingSnapshot,omitempty"`
	RequestFingerprint          string             `json:"requestFingerprint,omitempty"`
	UpstreamSubmissionStatus    string             `json:"upstreamSubmissionStatus"`
	UpstreamSubmissionUpdatedAt string             `json:"upstreamSubmissionUpdatedAt,omitempty"`
	ErrorCode                   string             `json:"errorCode,omitempty"`
	ErrorMessage                string             `json:"errorMessage,omitempty"`
	PublicError                 *publicerror.Error `json:"publicError,omitempty"`
	StartedAt                   time.Time          `json:"startedAt"`
	FinishedAt                  string             `json:"finishedAt,omitempty"`
}

type TaskParamPreprocessingLog struct {
	ID              string         `json:"id"`
	TaskID          string         `json:"taskId"`
	AttemptID       string         `json:"attemptId,omitempty"`
	AttemptNo       int            `json:"attemptNo,omitempty"`
	ModelType       string         `json:"modelType,omitempty"`
	PlatformID      string         `json:"platformId,omitempty"`
	PlatformModelID string         `json:"platformModelId,omitempty"`
	ClientID        string         `json:"clientId,omitempty"`
	Changed         bool           `json:"changed"`
	ChangeCount     int            `json:"changeCount"`
	ActualInput     map[string]any `json:"actualInput,omitempty"`
	ConvertedOutput map[string]any `json:"convertedOutput,omitempty"`
	Changes         []any          `json:"changes,omitempty"`
	ModelSnapshot   map[string]any `json:"model,omitempty"`
	CreatedAt       time.Time      `json:"createdAt"`
}

func (s *Store) ListPlatforms(ctx context.Context) ([]Platform, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, provider, platform_key, name, COALESCE(internal_name, ''), COALESCE(base_url, ''), auth_type, status,
       priority, dynamic_priority, COALESCE(dynamic_priority, priority),
       default_pricing_mode, default_discount_factor::float8, COALESCE(pricing_rule_set_id::text, ''),
       config, credentials, retry_policy, rate_limit_policy,
       COALESCE(to_char(cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''),
       created_at, updated_at
FROM integration_platforms
ORDER BY COALESCE(dynamic_priority, priority) ASC, priority ASC, created_at DESC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	platforms := make([]Platform, 0)
	for rows.Next() {
		var platform Platform
		var configBytes []byte
		var credentialsBytes []byte
		var retryPolicyBytes []byte
		var rateLimitPolicyBytes []byte
		var dynamicPriority sql.NullInt64
		if err := rows.Scan(
			&platform.ID,
			&platform.Provider,
			&platform.PlatformKey,
			&platform.Name,
			&platform.InternalName,
			&platform.BaseURL,
			&platform.AuthType,
			&platform.Status,
			&platform.Priority,
			&dynamicPriority,
			&platform.EffectivePriority,
			&platform.DefaultPricingMode,
			&platform.DefaultDiscountFactor,
			&platform.PricingRuleSetID,
			&configBytes,
			&credentialsBytes,
			&retryPolicyBytes,
			&rateLimitPolicyBytes,
			&platform.CooldownUntil,
			&platform.CreatedAt,
			&platform.UpdatedAt,
		); err != nil {
			return nil, err
		}
		platform.DynamicPriority = intPointerFromNull(dynamicPriority)
		platform.Config = decodeObject(configBytes)
		platform.CredentialsPreview = maskCredentialsPreview(credentialsBytes)
		platform.RetryPolicy = decodeObject(retryPolicyBytes)
		platform.RateLimitPolicy = decodeObject(rateLimitPolicyBytes)
		platforms = append(platforms, platform)
	}
	return platforms, rows.Err()
}

func (s *Store) CreatePlatform(ctx context.Context, input CreatePlatformInput) (Platform, error) {
	normalizedPolicy, err := NormalizeAndValidateRateLimitPolicy(input.RateLimitPolicy)
	if err != nil {
		return Platform{}, err
	}
	input.RateLimitPolicy = normalizedPolicy
	credentials, _ := json.Marshal(emptyObjectIfNil(input.Credentials))
	config, _ := json.Marshal(emptyObjectIfNil(input.Config))
	retryPolicy, _ := json.Marshal(emptyObjectIfNil(input.RetryPolicy))
	rateLimitPolicy, _ := json.Marshal(emptyObjectIfNil(input.RateLimitPolicy))
	if input.DefaultPricingMode == "" {
		input.DefaultPricingMode = "inherit_discount"
	}
	if input.DefaultDiscountFactor == 0 {
		input.DefaultDiscountFactor = 1
	}
	if input.Priority == 0 {
		input.Priority = 100
	}
	var platform Platform
	var configBytes []byte
	var credentialsResultBytes []byte
	var retryPolicyBytes []byte
	var rateLimitPolicyBytes []byte
	var dynamicPriority sql.NullInt64
	err = s.pool.QueryRow(ctx, `
INSERT INTO integration_platforms (
  provider, platform_key, name, internal_name, base_url, auth_type, credentials, config,
  default_pricing_mode, default_discount_factor, pricing_rule_set_id,
  priority, retry_policy, rate_limit_policy, status
)
VALUES (
  $1, COALESCE(NULLIF($2, ''), 'platform_' || replace(gen_random_uuid()::text, '-', '')), $3, NULLIF($4, ''), $5, $6, $7, $8,
  $9, $10, NULLIF($11, '')::uuid, $12, $13, $14, COALESCE(NULLIF($15, ''), 'enabled')
)
RETURNING id::text, provider, platform_key, name, COALESCE(internal_name, ''), COALESCE(base_url, ''), auth_type, status,
          priority, dynamic_priority, COALESCE(dynamic_priority, priority),
          default_pricing_mode, default_discount_factor::float8, COALESCE(pricing_rule_set_id::text, ''),
          config, credentials, retry_policy, rate_limit_policy,
          COALESCE(to_char(cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''),
          created_at, updated_at`,
		input.Provider, input.PlatformKey, input.Name, strings.TrimSpace(input.InternalName), input.BaseURL, input.AuthType, credentials, config,
		input.DefaultPricingMode, input.DefaultDiscountFactor, input.PricingRuleSetID, input.Priority,
		string(retryPolicy), string(rateLimitPolicy),
		input.Status,
	).Scan(
		&platform.ID,
		&platform.Provider,
		&platform.PlatformKey,
		&platform.Name,
		&platform.InternalName,
		&platform.BaseURL,
		&platform.AuthType,
		&platform.Status,
		&platform.Priority,
		&dynamicPriority,
		&platform.EffectivePriority,
		&platform.DefaultPricingMode,
		&platform.DefaultDiscountFactor,
		&platform.PricingRuleSetID,
		&configBytes,
		&credentialsResultBytes,
		&retryPolicyBytes,
		&rateLimitPolicyBytes,
		&platform.CooldownUntil,
		&platform.CreatedAt,
		&platform.UpdatedAt,
	)
	if err != nil {
		return Platform{}, err
	}
	platform.DynamicPriority = intPointerFromNull(dynamicPriority)
	platform.Config = decodeObject(configBytes)
	platform.CredentialsPreview = maskCredentialsPreview(credentialsResultBytes)
	platform.RetryPolicy = decodeObject(retryPolicyBytes)
	platform.RateLimitPolicy = decodeObject(rateLimitPolicyBytes)
	return platform, nil
}

func (s *Store) UpdatePlatform(ctx context.Context, id string, input CreatePlatformInput) (Platform, error) {
	normalizedPolicy, err := NormalizeAndValidateRateLimitPolicy(input.RateLimitPolicy)
	if err != nil {
		return Platform{}, err
	}
	input.RateLimitPolicy = normalizedPolicy
	var credentials any
	if input.Credentials != nil {
		credentialsBytes, _ := json.Marshal(emptyObjectIfNil(input.Credentials))
		credentials = string(credentialsBytes)
	}
	config, _ := json.Marshal(emptyObjectIfNil(input.Config))
	retryPolicy, _ := json.Marshal(emptyObjectIfNil(input.RetryPolicy))
	rateLimitPolicy, _ := json.Marshal(emptyObjectIfNil(input.RateLimitPolicy))
	if input.DefaultPricingMode == "" {
		input.DefaultPricingMode = "inherit_discount"
	}
	if input.DefaultDiscountFactor == 0 {
		input.DefaultDiscountFactor = 1
	}
	if input.Priority == 0 {
		input.Priority = 100
	}
	var platform Platform
	var configBytes []byte
	var credentialsResultBytes []byte
	var retryPolicyBytes []byte
	var rateLimitPolicyBytes []byte
	var dynamicPriority sql.NullInt64
	err = s.pool.QueryRow(ctx, `
UPDATE integration_platforms
SET provider = $2,
    platform_key = COALESCE(NULLIF($3, ''), platform_key),
    name = $4,
    internal_name = NULLIF($5, ''),
    base_url = $6,
    auth_type = $7,
    credentials = CASE
      WHEN $8::jsonb IS NULL THEN credentials
      WHEN $8::jsonb = '{}'::jsonb THEN '{}'::jsonb
      ELSE credentials || $8::jsonb
    END,
    config = $9,
    default_pricing_mode = $10,
    default_discount_factor = $11,
    pricing_rule_set_id = NULLIF($12, '')::uuid,
    priority = $13,
    retry_policy = $14,
    rate_limit_policy = $15,
    status = COALESCE(NULLIF($16, ''), status),
    updated_at = now()
WHERE id = $1::uuid
RETURNING id::text, provider, platform_key, name, COALESCE(internal_name, ''), COALESCE(base_url, ''), auth_type, status,
          priority, dynamic_priority, COALESCE(dynamic_priority, priority),
          default_pricing_mode, default_discount_factor::float8, COALESCE(pricing_rule_set_id::text, ''),
          config, credentials, retry_policy, rate_limit_policy,
          COALESCE(to_char(cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''),
          created_at, updated_at`,
		id,
		input.Provider,
		input.PlatformKey,
		input.Name,
		strings.TrimSpace(input.InternalName),
		input.BaseURL,
		input.AuthType,
		credentials,
		string(config),
		input.DefaultPricingMode,
		input.DefaultDiscountFactor,
		input.PricingRuleSetID,
		input.Priority,
		string(retryPolicy),
		string(rateLimitPolicy),
		input.Status,
	).Scan(
		&platform.ID,
		&platform.Provider,
		&platform.PlatformKey,
		&platform.Name,
		&platform.InternalName,
		&platform.BaseURL,
		&platform.AuthType,
		&platform.Status,
		&platform.Priority,
		&dynamicPriority,
		&platform.EffectivePriority,
		&platform.DefaultPricingMode,
		&platform.DefaultDiscountFactor,
		&platform.PricingRuleSetID,
		&configBytes,
		&credentialsResultBytes,
		&retryPolicyBytes,
		&rateLimitPolicyBytes,
		&platform.CooldownUntil,
		&platform.CreatedAt,
		&platform.UpdatedAt,
	)
	if err != nil {
		return Platform{}, err
	}
	platform.DynamicPriority = intPointerFromNull(dynamicPriority)
	platform.Config = decodeObject(configBytes)
	platform.CredentialsPreview = maskCredentialsPreview(credentialsResultBytes)
	platform.RetryPolicy = decodeObject(retryPolicyBytes)
	platform.RateLimitPolicy = decodeObject(rateLimitPolicyBytes)
	return platform, nil
}

func (s *Store) DeletePlatform(ctx context.Context, id string) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)

	rows, err := tx.Query(ctx, `SELECT id::text FROM platform_models WHERE platform_id = $1::uuid`, id)
	if err != nil {
		return err
	}
	modelIDs := make([]string, 0)
	for rows.Next() {
		var modelID string
		if err := rows.Scan(&modelID); err != nil {
			rows.Close()
			return err
		}
		modelIDs = append(modelIDs, modelID)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return err
	}
	rows.Close()

	result, err := tx.Exec(ctx, `DELETE FROM integration_platforms WHERE id = $1::uuid`, id)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	if _, err := tx.Exec(ctx, `
DELETE FROM gateway_access_rules
WHERE (resource_type = 'platform' AND resource_id = $1::uuid)
   OR (resource_type = 'platform_model' AND resource_id::text = ANY($2::text[]))`, id, modelIDs); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) ListModels(ctx context.Context) ([]PlatformModel, error) {
	return s.listModels(ctx, "")
}

func (s *Store) ListPlatformModels(ctx context.Context, platformID string) ([]PlatformModel, error) {
	return s.listModels(ctx, strings.TrimSpace(platformID))
}

func (s *Store) listModels(ctx context.Context, platformID string) ([]PlatformModel, error) {
	args := []any{}
	where := ""
	if platformID != "" {
		where = "WHERE m.platform_id = $1::uuid"
		args = append(args, platformID)
	}

	rows, err := s.pool.Query(ctx, `
SELECT m.id::text, m.platform_id::text, COALESCE(m.base_model_id::text, ''), p.provider, p.name,
       COALESCE(NULLIF(b.invocation_name, ''), m.model_name),
       COALESCE(NULLIF(m.provider_model_name, ''), m.model_name),
       COALESCE(NULLIF(b.invocation_name, ''), NULLIF(m.model_alias, ''), m.model_name),
       COALESCE(b.model_type, m.model_type),
       COALESCE(NULLIF(b.display_name, ''), NULLIF(m.display_name, ''), COALESCE(NULLIF(b.invocation_name, ''), m.model_name)),
       COALESCE(b.legacy_aliases, '[]'::jsonb),
       m.capability_override, m.capabilities, COALESCE(b.capabilities, '{}'::jsonb),
       COALESCE(b.base_billing_config, '{}'::jsonb), COALESCE(b.pricing_rule_set_id::text, ''),
       COALESCE(p.pricing_rule_set_id::text, ''), m.pricing_mode, COALESCE(m.discount_factor, 0)::float8,
       COALESCE(m.pricing_rule_set_id::text, ''), m.billing_config_override, m.billing_config,
       m.permission_config, m.retry_policy, m.rate_limit_policy, m.rate_limit_policy_mode,
       COALESCE(m.runtime_policy_set_id::text, ''), m.runtime_policy_override,
       COALESCE(b.default_rate_limit_policy, '{}'::jsonb), COALESCE(b.runtime_policy_set_id::text, ''),
       COALESCE(to_char(m.cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''),
       m.enabled, m.created_at, m.updated_at
FROM platform_models m
JOIN integration_platforms p ON p.id = m.platform_id
LEFT JOIN LATERAL (
  SELECT catalog.invocation_name, catalog.display_name, catalog.model_type,
         catalog.capabilities, catalog.base_billing_config, catalog.pricing_rule_set_id,
         catalog.default_rate_limit_policy, catalog.runtime_policy_set_id,
         COALESCE((
           SELECT jsonb_agg(DISTINCT compatibility_alias.alias ORDER BY compatibility_alias.alias)
           FROM model_compatibility_aliases compatibility_alias
           WHERE compatibility_alias.base_model_id = catalog.id
             AND compatibility_alias.active = true
             AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now())
         ), '[]'::jsonb) AS legacy_aliases
  FROM base_model_catalog catalog
  WHERE (m.base_model_id IS NOT NULL AND catalog.id = m.base_model_id)
     OR (
       catalog.provider_key = p.provider
       AND (
         catalog.provider_model_name = COALESCE(NULLIF(m.provider_model_name, ''), m.model_name)
         OR catalog.canonical_model_key = p.provider || ':' || COALESCE(NULLIF(m.provider_model_name, ''), m.model_name)
       )
     )
  ORDER BY CASE WHEN m.base_model_id IS NOT NULL AND catalog.id = m.base_model_id THEN 0 ELSE 1 END,
           catalog.updated_at DESC
  LIMIT 1
) b ON true
`+where+`
ORDER BY m.model_type ASC, m.model_name ASC`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	models := make([]PlatformModel, 0)
	for rows.Next() {
		var model PlatformModel
		var capabilityOverride []byte
		var capabilities []byte
		var baseCapabilities []byte
		var baseBillingConfig []byte
		var billingConfigOverride []byte
		var billingConfig []byte
		var permissionConfig []byte
		var retryPolicy []byte
		var rateLimitPolicy []byte
		var runtimePolicyOverride []byte
		var baseRateLimitPolicy []byte
		var modelTypeBytes []byte
		var legacyAliasesBytes []byte
		if err := rows.Scan(
			&model.ID,
			&model.PlatformID,
			&model.BaseModelID,
			&model.Provider,
			&model.PlatformName,
			&model.ModelName,
			&model.ProviderModelName,
			&model.ModelAlias,
			&modelTypeBytes,
			&model.DisplayName,
			&legacyAliasesBytes,
			&capabilityOverride,
			&capabilities,
			&baseCapabilities,
			&baseBillingConfig,
			&model.BasePricingRuleSetID,
			&model.PlatformPricingRuleSetID,
			&model.PricingMode,
			&model.DiscountFactor,
			&model.PricingRuleSetID,
			&billingConfigOverride,
			&billingConfig,
			&permissionConfig,
			&retryPolicy,
			&rateLimitPolicy,
			&model.RateLimitPolicyMode,
			&model.RuntimePolicySetID,
			&runtimePolicyOverride,
			&baseRateLimitPolicy,
			&model.BaseRuntimePolicySetID,
			&model.CooldownUntil,
			&model.Enabled,
			&model.CreatedAt,
			&model.UpdatedAt,
		); err != nil {
			return nil, err
		}
		model.CapabilityOverride = decodeObject(capabilityOverride)
		model.Capabilities = decodeObject(capabilities)
		model.BaseCapabilities = decodeObject(baseCapabilities)
		model.BaseBillingConfig = decodeObject(baseBillingConfig)
		model.ModelType = decodeStringArray(modelTypeBytes)
		model.LegacyAliases = decodeStringArray(legacyAliasesBytes)
		model.BillingConfigOverride = decodeObject(billingConfigOverride)
		model.BillingConfig = decodeObject(billingConfig)
		model.PermissionConfig = decodeObject(permissionConfig)
		model.RetryPolicy = decodeObject(retryPolicy)
		model.RateLimitPolicy = decodeObject(rateLimitPolicy)
		model.RuntimePolicyOverride = decodeObject(runtimePolicyOverride)
		model.BaseRateLimitPolicy = decodeObject(baseRateLimitPolicy)
		models = append(models, model)
	}
	return models, rows.Err()
}

func (s *Store) ListCatalogProviders(ctx context.Context) ([]CatalogProvider, error) {
	rows, err := s.pool.Query(ctx, `
SELECT `+catalogProviderColumns+`
FROM model_catalog_providers
ORDER BY provider_key ASC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]CatalogProvider, 0)
	for rows.Next() {
		item, err := scanCatalogProvider(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListPricingRules(ctx context.Context) ([]PricingRule, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, scope_type, COALESCE(scope_id::text, ''), resource_type, unit,
       base_price::float8, currency, base_weight, dynamic_weight,
       COALESCE(rule_set_id::text, ''), rule_key, display_name, calculator_type,
       dimension_schema, formula_config, priority, status, metadata, created_at, updated_at
FROM model_pricing_rules
ORDER BY COALESCE(rule_set_id::text, ''), priority ASC, resource_type ASC, created_at DESC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]PricingRule, 0)
	for rows.Next() {
		var item PricingRule
		var baseWeight []byte
		var dynamicWeight []byte
		var dimensionSchema []byte
		var formulaConfig []byte
		var metadata []byte
		if err := rows.Scan(
			&item.ID,
			&item.ScopeType,
			&item.ScopeID,
			&item.ResourceType,
			&item.Unit,
			&item.BasePrice,
			&item.Currency,
			&baseWeight,
			&dynamicWeight,
			&item.RuleSetID,
			&item.RuleKey,
			&item.DisplayName,
			&item.CalculatorType,
			&dimensionSchema,
			&formulaConfig,
			&item.Priority,
			&item.Status,
			&metadata,
			&item.CreatedAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, err
		}
		item.BaseWeight = decodeObject(baseWeight)
		item.DynamicWeight = decodeObject(dynamicWeight)
		item.DimensionSchema = decodeObject(dimensionSchema)
		item.FormulaConfig = decodeObject(formulaConfig)
		item.Metadata = decodeObject(metadata)
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListTenants(ctx context.Context) ([]GatewayTenant, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, tenant_key, source, COALESCE(external_tenant_id, ''), name, COALESCE(description, ''),
       COALESCE(default_user_group_id::text, ''), COALESCE(plan_key, ''), billing_profile, rate_limit_policy,
       auth_policy, metadata, status, COALESCE(synced_at::text, ''), COALESCE(source_updated_at::text, ''),
       created_at, updated_at
FROM gateway_tenants
WHERE deleted_at IS NULL
ORDER BY created_at DESC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]GatewayTenant, 0)
	for rows.Next() {
		var item GatewayTenant
		var billingProfile []byte
		var rateLimitPolicy []byte
		var authPolicy []byte
		var metadata []byte
		if err := rows.Scan(
			&item.ID,
			&item.TenantKey,
			&item.Source,
			&item.ExternalTenantID,
			&item.Name,
			&item.Description,
			&item.DefaultUserGroupID,
			&item.PlanKey,
			&billingProfile,
			&rateLimitPolicy,
			&authPolicy,
			&metadata,
			&item.Status,
			&item.SyncedAt,
			&item.SourceUpdatedAt,
			&item.CreatedAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, err
		}
		item.BillingProfile = decodeObject(billingProfile)
		item.RateLimitPolicy = decodeObject(rateLimitPolicy)
		item.AuthPolicy = decodeObject(authPolicy)
		item.Metadata = decodeObject(metadata)
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListUsers(ctx context.Context) ([]GatewayUser, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, user_key, source, COALESCE(external_user_id, ''), username,
       COALESCE(display_name, ''), COALESCE(email, ''), COALESCE(phone, ''), COALESCE(avatar_url, ''),
       COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''),
       COALESCE(default_user_group_id::text, ''), roles, auth_profile, metadata,
       status, COALESCE(last_login_at::text, ''), COALESCE(synced_at::text, ''), COALESCE(source_updated_at::text, ''),
       created_at, updated_at, COALESCE(wallets.wallet_accounts, '[]'::jsonb)
FROM gateway_users u
LEFT JOIN LATERAL (
  SELECT jsonb_agg(jsonb_build_object(
    'id', a.id::text,
    'gatewayTenantId', COALESCE(a.gateway_tenant_id::text, ''),
    'gatewayUserId', a.gateway_user_id::text,
    'tenantId', COALESCE(a.tenant_id, ''),
    'tenantKey', COALESCE(a.tenant_key, ''),
    'userId', COALESCE(a.user_id, ''),
    'currency', a.currency,
    'balance', a.balance::float8,
    'frozenBalance', a.frozen_balance::float8,
    'totalRecharged', a.total_recharged::float8,
    'totalSpent', a.total_spent::float8,
    'status', a.status,
    'metadata', a.metadata,
    'createdAt', a.created_at,
    'updatedAt', a.updated_at
  ) ORDER BY a.currency ASC) AS wallet_accounts
  FROM gateway_wallet_accounts a
  WHERE a.gateway_user_id = u.id
) wallets ON true
WHERE u.deleted_at IS NULL
ORDER BY u.created_at DESC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]GatewayUser, 0)
	for rows.Next() {
		var item GatewayUser
		var roles []byte
		var authProfile []byte
		var metadata []byte
		var walletAccounts []byte
		if err := rows.Scan(
			&item.ID,
			&item.UserKey,
			&item.Source,
			&item.ExternalUserID,
			&item.Username,
			&item.DisplayName,
			&item.Email,
			&item.Phone,
			&item.AvatarURL,
			&item.GatewayTenantID,
			&item.TenantID,
			&item.TenantKey,
			&item.DefaultUserGroupID,
			&roles,
			&authProfile,
			&metadata,
			&item.Status,
			&item.LastLoginAt,
			&item.SyncedAt,
			&item.SourceUpdatedAt,
			&item.CreatedAt,
			&item.UpdatedAt,
			&walletAccounts,
		); err != nil {
			return nil, err
		}
		item.Roles = decodeStringArray(roles)
		item.AuthProfile = decodeObject(authProfile)
		item.Metadata = decodeObject(metadata)
		item.WalletAccounts = decodeWalletAccounts(walletAccounts)
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListUserGroups(ctx context.Context) ([]UserGroup, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, group_key, name, COALESCE(description, ''), source, priority,
       recharge_discount_policy, billing_discount_policy, rate_limit_policy, quota_policy, metadata,
       status, created_at, updated_at
FROM gateway_user_groups
ORDER BY priority ASC, group_key ASC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]UserGroup, 0)
	for rows.Next() {
		var item UserGroup
		var rechargeDiscountPolicy []byte
		var billingDiscountPolicy []byte
		var rateLimitPolicy []byte
		var quotaPolicy []byte
		var metadata []byte
		if err := rows.Scan(
			&item.ID,
			&item.GroupKey,
			&item.Name,
			&item.Description,
			&item.Source,
			&item.Priority,
			&rechargeDiscountPolicy,
			&billingDiscountPolicy,
			&rateLimitPolicy,
			&quotaPolicy,
			&metadata,
			&item.Status,
			&item.CreatedAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, err
		}
		item.RechargeDiscountPolicy = decodeObject(rechargeDiscountPolicy)
		item.BillingDiscountPolicy = decodeObject(billingDiscountPolicy)
		item.RateLimitPolicy = decodeObject(rateLimitPolicy)
		item.QuotaPolicy = decodeObject(quotaPolicy)
		item.Metadata = decodeObject(metadata)
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListUserGroupsBySubject(ctx context.Context, groupIDs []string, groupKeys []string) ([]UserGroup, error) {
	rows, err := s.pool.Query(ctx, `
SELECT `+userGroupColumns+`
FROM gateway_user_groups
WHERE status = 'active'
  AND (
    id::text = ANY($1::text[])
    OR group_key = ANY($2::text[])
  )
ORDER BY priority ASC, group_key ASC`, groupIDs, groupKeys)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]UserGroup, 0)
	for rows.Next() {
		item, err := scanUserGroup(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListAPIKeys(ctx context.Context, user *auth.User) ([]APIKey, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return []APIKey{}, nil
	}
	rows, err := s.pool.Query(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       key_prefix, name, scopes, COALESCE(user_group_id::text, ''),
       rate_limit_policy, quota_policy, status, COALESCE(expires_at::text, ''),
       COALESCE(last_used_at::text, ''), created_at, updated_at,
       CASE WHEN status = 'active' AND (expires_at IS NULL OR expires_at > now())
            THEN COALESCE(key_secret, '')
            ELSE ''
       END
FROM gateway_api_keys
WHERE gateway_user_id = $1::uuid AND deleted_at IS NULL
ORDER BY created_at DESC`, gatewayUserID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]APIKey, 0)
	for rows.Next() {
		var secret string
		item, err := scanAPIKeyWithSecret(rows, &secret)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) ListPlayableAPIKeys(ctx context.Context, user *auth.User) ([]PlayableAPIKey, error) {
	items, err := s.listRecoverableAPIKeys(ctx, user)
	if err != nil {
		return nil, err
	}
	if len(items) > 0 {
		return items, nil
	}
	created, err := s.CreateAPIKey(ctx, CreateAPIKeyInput{
		Name:   "Playground API Key",
		Scopes: defaultAPIKeyScopes(),
	}, user)
	if err != nil {
		return nil, err
	}
	return []PlayableAPIKey{{APIKey: created.APIKey, Secret: created.Secret}}, nil
}

func (s *Store) listRecoverableAPIKeys(ctx context.Context, user *auth.User) ([]PlayableAPIKey, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return nil, ErrLocalUserRequired
	}
	rows, err := s.pool.Query(ctx, `
SELECT id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       key_prefix, name, scopes, COALESCE(user_group_id::text, ''),
       rate_limit_policy, quota_policy, status, COALESCE(expires_at::text, ''),
       COALESCE(last_used_at::text, ''), created_at, updated_at, COALESCE(key_secret, '')
FROM gateway_api_keys
WHERE gateway_user_id = $1::uuid
  AND status = 'active'
  AND deleted_at IS NULL
  AND COALESCE(key_secret, '') <> ''
  AND (expires_at IS NULL OR expires_at > now())
ORDER BY created_at DESC`, gatewayUserID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]PlayableAPIKey, 0)
	for rows.Next() {
		var item PlayableAPIKey
		apiKey, err := scanAPIKeyWithSecret(rows, &item.Secret)
		if err != nil {
			return nil, err
		}
		item.APIKey = apiKey
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) CreateAPIKey(ctx context.Context, input CreateAPIKeyInput, user *auth.User) (CreatedAPIKey, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return CreatedAPIKey{}, ErrLocalUserRequired
	}
	name := strings.TrimSpace(input.Name)
	if name == "" {
		name = "Default API Key"
	}
	scopes := input.Scopes
	if len(scopes) == 0 {
		scopes = defaultAPIKeyScopes()
	}
	scopes = normalizeAPIKeyScopes(scopes)
	secret, err := generateAPIKeySecret()
	if err != nil {
		return CreatedAPIKey{}, err
	}
	keyHash, err := bcrypt.GenerateFromPassword([]byte(secret), bcrypt.DefaultCost)
	if err != nil {
		return CreatedAPIKey{}, err
	}
	scopesJSON, err := json.Marshal(scopes)
	if err != nil {
		return CreatedAPIKey{}, err
	}

	var item APIKey
	var scopesBytes []byte
	var rateLimitPolicy []byte
	var quotaPolicy []byte
	err = s.pool.QueryRow(ctx, `
INSERT INTO gateway_api_keys (
  gateway_tenant_id, gateway_user_id, tenant_id, tenant_key, user_id,
  key_prefix, key_secret, key_hash, name, scopes, expires_at
)
VALUES (NULLIF($1, '')::uuid, $2::uuid, NULLIF($3, ''), NULLIF($4, ''), NULLIF($5, ''),
        $6, $7, $8, $9, $10::jsonb, NULLIF($11, '')::timestamptz)
RETURNING id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       key_prefix, name, scopes, COALESCE(user_group_id::text, ''),
       rate_limit_policy, quota_policy, status, COALESCE(expires_at::text, ''),
       COALESCE(last_used_at::text, ''), created_at, updated_at`,
		user.GatewayTenantID, gatewayUserID, user.TenantID, user.TenantKey, user.ID,
		apiKeyPrefix(secret), secret, string(keyHash), name, string(scopesJSON), strings.TrimSpace(input.ExpiresAt),
	).Scan(
		&item.ID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.TenantID,
		&item.TenantKey,
		&item.UserID,
		&item.KeyPrefix,
		&item.Name,
		&scopesBytes,
		&item.UserGroupID,
		&rateLimitPolicy,
		&quotaPolicy,
		&item.Status,
		&item.ExpiresAt,
		&item.LastUsedAt,
		&item.CreatedAt,
		&item.UpdatedAt,
	)
	if err != nil {
		return CreatedAPIKey{}, err
	}
	item.Scopes = decodeStringArray(scopesBytes)
	item.RateLimitPolicy = decodeObject(rateLimitPolicy)
	item.QuotaPolicy = decodeObject(quotaPolicy)
	return CreatedAPIKey{APIKey: item, Secret: secret}, nil
}

func (s *Store) UpdateAPIKeyScopes(ctx context.Context, apiKeyID string, input UpdateAPIKeyScopesInput, user *auth.User) (APIKey, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return APIKey{}, ErrLocalUserRequired
	}
	scopes := normalizeAPIKeyScopes(input.Scopes)
	if len(scopes) == 0 {
		return APIKey{}, ErrInvalidAPIKeyScopes
	}
	scopesJSON, err := json.Marshal(scopes)
	if err != nil {
		return APIKey{}, err
	}
	return scanAPIKey(s.pool.QueryRow(ctx, `
UPDATE gateway_api_keys
SET scopes = $3::jsonb, updated_at = now()
WHERE id = $1::uuid AND gateway_user_id = $2::uuid AND deleted_at IS NULL
RETURNING id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       key_prefix, name, scopes, COALESCE(user_group_id::text, ''),
       rate_limit_policy, quota_policy, status, COALESCE(expires_at::text, ''),
       COALESCE(last_used_at::text, ''), created_at, updated_at`,
		apiKeyID, gatewayUserID, string(scopesJSON),
	))
}

func (s *Store) DisableAPIKey(ctx context.Context, apiKeyID string, user *auth.User) (APIKey, error) {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return APIKey{}, ErrLocalUserRequired
	}
	var item APIKey
	var scopesBytes []byte
	var rateLimitPolicy []byte
	var quotaPolicy []byte
	err := s.pool.QueryRow(ctx, `
UPDATE gateway_api_keys
SET status = 'disabled', updated_at = now()
WHERE id = $1::uuid AND gateway_user_id = $2::uuid AND deleted_at IS NULL
RETURNING id::text, COALESCE(gateway_tenant_id::text, ''), gateway_user_id::text,
       COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(user_id, ''),
       key_prefix, name, scopes, COALESCE(user_group_id::text, ''),
       rate_limit_policy, quota_policy, status, COALESCE(expires_at::text, ''),
       COALESCE(last_used_at::text, ''), created_at, updated_at`,
		apiKeyID, gatewayUserID,
	).Scan(
		&item.ID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.TenantID,
		&item.TenantKey,
		&item.UserID,
		&item.KeyPrefix,
		&item.Name,
		&scopesBytes,
		&item.UserGroupID,
		&rateLimitPolicy,
		&quotaPolicy,
		&item.Status,
		&item.ExpiresAt,
		&item.LastUsedAt,
		&item.CreatedAt,
		&item.UpdatedAt,
	)
	if err != nil {
		return APIKey{}, err
	}
	item.Scopes = decodeStringArray(scopesBytes)
	item.RateLimitPolicy = decodeObject(rateLimitPolicy)
	item.QuotaPolicy = decodeObject(quotaPolicy)
	return item, nil
}

func (s *Store) DeleteAPIKey(ctx context.Context, apiKeyID string, user *auth.User) error {
	gatewayUserID := localGatewayUserID(user)
	if gatewayUserID == "" {
		return ErrLocalUserRequired
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)

	result, err := tx.Exec(ctx, `
UPDATE gateway_api_keys
SET status = 'revoked',
    key_secret = NULL,
    deleted_at = now(),
    updated_at = now()
WHERE id = $1::uuid AND gateway_user_id = $2::uuid AND deleted_at IS NULL`, apiKeyID, gatewayUserID)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	if _, err := tx.Exec(ctx, `
DELETE FROM gateway_access_rules
WHERE subject_type = 'api_key' AND subject_id = $1::uuid`, apiKeyID); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) VerifyLocalAPIKey(ctx context.Context, secret string) (*auth.User, error) {
	return verifyLocalAPIKey(ctx, s.pool, secret)
}

type localAPIKeyDatabase interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}

type localAPIKeyCandidate struct {
	apiKeyID        string
	hash            string
	keyPrefix       string
	keyName         string
	scopesBytes     []byte
	userGroupID     string
	gatewayUserID   string
	username        string
	rolesBytes      []byte
	gatewayTenantID string
	tenantID        string
	tenantKey       string
}

func verifyLocalAPIKey(ctx context.Context, database localAPIKeyDatabase, secret string) (*auth.User, error) {
	prefix := apiKeyPrefix(secret)
	if prefix == "" {
		return nil, auth.ErrUnauthorized
	}
	rows, err := database.Query(ctx, `
SELECT k.id::text, k.key_hash, k.key_prefix, k.name, k.scopes, COALESCE(k.user_group_id::text, u.default_user_group_id::text, ''),
       u.id::text, u.username, u.roles, COALESCE(u.gateway_tenant_id::text, ''),
       COALESCE(u.tenant_id, ''), COALESCE(u.tenant_key, '')
FROM gateway_api_keys k
JOIN gateway_users u ON u.id = k.gateway_user_id
WHERE k.key_prefix = $1
  AND k.status = 'active'
  AND k.deleted_at IS NULL
  AND u.status = 'active'
  AND u.deleted_at IS NULL
  AND (k.expires_at IS NULL OR k.expires_at > now())`, prefix)
	if err != nil {
		return nil, err
	}
	candidates, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (localAPIKeyCandidate, error) {
		var candidate localAPIKeyCandidate
		err := row.Scan(
			&candidate.apiKeyID,
			&candidate.hash,
			&candidate.keyPrefix,
			&candidate.keyName,
			&candidate.scopesBytes,
			&candidate.userGroupID,
			&candidate.gatewayUserID,
			&candidate.username,
			&candidate.rolesBytes,
			&candidate.gatewayTenantID,
			&candidate.tenantID,
			&candidate.tenantKey,
		)
		return candidate, err
	})
	if err != nil {
		return nil, err
	}

	for _, candidate := range candidates {
		if bcrypt.CompareHashAndPassword([]byte(candidate.hash), []byte(secret)) != nil {
			continue
		}
		// API key usage timestamps are telemetry, not authentication state. Keep
		// their writes off the synchronous-replication critical path and coalesce
		// hot-key updates so concurrent requests do not contend on the same row.
		if _, err := database.Exec(ctx, `
WITH commit_policy AS (
  SELECT set_config('synchronous_commit', 'off', true)
)
UPDATE gateway_api_keys
SET last_used_at = now(),
    updated_at = now()
FROM commit_policy
WHERE id = $1::uuid
  AND (last_used_at IS NULL OR last_used_at < now() - interval '1 minute')`, candidate.apiKeyID); err != nil {
			return nil, err
		}
		return &auth.User{
			ID:              candidate.gatewayUserID,
			Username:        candidate.username,
			Roles:           decodeStringArray(candidate.rolesBytes),
			TenantID:        candidate.tenantID,
			GatewayTenantID: candidate.gatewayTenantID,
			TenantKey:       candidate.tenantKey,
			Source:          "gateway",
			GatewayUserID:   candidate.gatewayUserID,
			UserGroupID:     candidate.userGroupID,
			APIKeyID:        candidate.apiKeyID,
			APIKeyName:      candidate.keyName,
			APIKeyPrefix:    candidate.keyPrefix,
			APIKeyScopes:    decodeStringArray(candidate.scopesBytes),
		}, nil
	}
	return nil, auth.ErrUnauthorized
}

func (s *Store) RegisterLocalUser(ctx context.Context, input LocalRegisterInput) (GatewayUser, error) {
	account := normalizeAccount(firstNonEmpty(input.Username, input.Email))
	if account == "" {
		return GatewayUser{}, errors.New("username or email is required")
	}
	if len(input.Password) < 8 {
		return GatewayUser{}, ErrWeakPassword
	}
	tenantKey := "default"
	tenantName := "Default Tenant"
	displayName := strings.TrimSpace(input.DisplayName)
	username := strings.TrimSpace(input.Username)
	if username == "" {
		username = account
	}
	email := strings.TrimSpace(strings.ToLower(input.Email))
	invitationCode := strings.TrimSpace(input.InvitationCode)

	passwordHash, err := bcrypt.GenerateFromPassword([]byte(input.Password), bcrypt.DefaultCost)
	if err != nil {
		return GatewayUser{}, err
	}

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return GatewayUser{}, err
	}
	defer rollbackTransaction(tx)

	var tenantID string
	userGroupID := ""
	role := "user"
	invitationID := ""
	invitedBy := ""
	var isBootstrapUser bool
	if err := tx.QueryRow(ctx, `
SELECT NOT EXISTS (
  SELECT 1 FROM gateway_users WHERE source = 'gateway' AND deleted_at IS NULL
)`).Scan(&isBootstrapUser); err != nil {
		return GatewayUser{}, err
	}
	if isBootstrapUser {
		role = "manager"
	}
	if err := tx.QueryRow(ctx, `
		INSERT INTO gateway_tenants (tenant_key, source, external_tenant_id, name)
		VALUES ($1, 'gateway', $1, $2)
		ON CONFLICT (tenant_key) DO UPDATE SET updated_at=now()
		RETURNING id::text`,
		tenantKey, tenantName,
	).Scan(&tenantID); err != nil {
		return GatewayUser{}, err
	}
	_ = tx.QueryRow(ctx, `
SELECT COALESCE(default_user_group_id::text, '')
FROM gateway_tenants
WHERE id = $1::uuid`, tenantID).Scan(&userGroupID)
	if userGroupID == "" {
		_ = tx.QueryRow(ctx, `
SELECT id::text
FROM gateway_user_groups
WHERE group_key = 'default' AND status = 'active'
LIMIT 1`).Scan(&userGroupID)
	}
	if invitationCode != "" {
		if err := tx.QueryRow(ctx, `
SELECT i.id::text, COALESCE(i.created_by::text, '')
FROM gateway_invitations i
WHERE lower(i.invite_code) = lower($1)
  AND i.status = 'active'
  AND (i.expires_at IS NULL OR i.expires_at > now())
  AND (i.max_uses IS NULL OR i.used_count < i.max_uses)
FOR UPDATE OF i`,
			invitationCode,
		).Scan(&invitationID, &invitedBy); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return GatewayUser{}, ErrInvalidInvitation
			}
			return GatewayUser{}, err
		}
	}

	rolesJSON, err := json.Marshal([]string{role})
	if err != nil {
		return GatewayUser{}, err
	}
	userMetadataJSON := []byte("{}")
	if invitationID != "" {
		userMetadataJSON, err = json.Marshal(map[string]any{
			"registration": map[string]any{
				"invitationId":   invitationID,
				"invitationCode": invitationCode,
				"invitedBy":      invitedBy,
			},
		})
		if err != nil {
			return GatewayUser{}, err
		}
	}

	var user GatewayUser
	var roles []byte
	var authProfile []byte
	var metadata []byte
	if err := tx.QueryRow(ctx, `
		INSERT INTO gateway_users (
		  user_key, source, external_user_id, username, display_name, email,
		  password_hash, gateway_tenant_id, tenant_id, tenant_key, default_user_group_id, roles, metadata, status
		)
		VALUES ($1, 'gateway', $2, $3, NULLIF($4, ''), NULLIF($5, ''), $6, $7::uuid, $8, $8, NULLIF($9, '')::uuid, $10::jsonb, $11::jsonb, 'active')
		RETURNING id::text, user_key, source, COALESCE(external_user_id, ''), username,
		       COALESCE(display_name, ''), COALESCE(email, ''), COALESCE(phone, ''), COALESCE(avatar_url, ''),
		       COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''),
		       COALESCE(default_user_group_id::text, ''), roles, auth_profile, metadata,
		       status, COALESCE(last_login_at::text, ''), COALESCE(synced_at::text, ''), COALESCE(source_updated_at::text, ''),
		       created_at, updated_at`,
		"gateway:"+account, account, username, displayName, email, string(passwordHash), tenantID, tenantKey, userGroupID, string(rolesJSON), string(userMetadataJSON),
	).Scan(
		&user.ID,
		&user.UserKey,
		&user.Source,
		&user.ExternalUserID,
		&user.Username,
		&user.DisplayName,
		&user.Email,
		&user.Phone,
		&user.AvatarURL,
		&user.GatewayTenantID,
		&user.TenantID,
		&user.TenantKey,
		&user.DefaultUserGroupID,
		&roles,
		&authProfile,
		&metadata,
		&user.Status,
		&user.LastLoginAt,
		&user.SyncedAt,
		&user.SourceUpdatedAt,
		&user.CreatedAt,
		&user.UpdatedAt,
	); err != nil {
		if isUniqueViolation(err) {
			return GatewayUser{}, ErrUserAlreadyExists
		}
		return GatewayUser{}, err
	}
	if invitationID != "" {
		if _, err := tx.Exec(ctx, `
UPDATE gateway_invitations
SET used_count = used_count + 1, updated_at = now()
WHERE id = $1::uuid`, invitationID); err != nil {
			return GatewayUser{}, err
		}
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO gateway_wallet_accounts (
  gateway_tenant_id, gateway_user_id, tenant_id, tenant_key, user_id, currency
)
VALUES ($1::uuid, $2::uuid, $3, $3, $4, 'resource')
ON CONFLICT (gateway_user_id, currency) DO NOTHING`,
		user.GatewayTenantID, user.ID, user.TenantKey, user.ID,
	); err != nil {
		return GatewayUser{}, err
	}
	if err := ensureWalletAccountAuditGuard(ctx, tx, user.ID, "resource"); err != nil {
		return GatewayUser{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return GatewayUser{}, err
	}
	user.Roles = decodeStringArray(roles)
	user.AuthProfile = decodeObject(authProfile)
	user.Metadata = decodeObject(metadata)
	return user, nil
}

func (s *Store) AuthenticateLocalUser(ctx context.Context, input LocalLoginInput) (GatewayUser, error) {
	account := normalizeAccount(input.Account)
	if account == "" || input.Password == "" {
		return GatewayUser{}, ErrInvalidCredentials
	}
	var user GatewayUser
	var passwordHash string
	var roles []byte
	var authProfile []byte
	var metadata []byte
	err := s.pool.QueryRow(ctx, `
SELECT id::text, user_key, source, COALESCE(external_user_id, ''), username,
       COALESCE(display_name, ''), COALESCE(email, ''), COALESCE(phone, ''), COALESCE(avatar_url, ''),
       COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''),
       COALESCE(default_user_group_id::text, ''), roles, auth_profile, metadata,
       status, COALESCE(password_hash, ''), COALESCE(last_login_at::text, ''), COALESCE(synced_at::text, ''),
       COALESCE(source_updated_at::text, ''), created_at, updated_at
FROM gateway_users
WHERE source='gateway'
  AND deleted_at IS NULL
  AND (external_user_id=$1 OR lower(username)=$1 OR lower(COALESCE(email, ''))=$1)
ORDER BY created_at ASC
LIMIT 1`, account,
	).Scan(
		&user.ID,
		&user.UserKey,
		&user.Source,
		&user.ExternalUserID,
		&user.Username,
		&user.DisplayName,
		&user.Email,
		&user.Phone,
		&user.AvatarURL,
		&user.GatewayTenantID,
		&user.TenantID,
		&user.TenantKey,
		&user.DefaultUserGroupID,
		&roles,
		&authProfile,
		&metadata,
		&user.Status,
		&passwordHash,
		&user.LastLoginAt,
		&user.SyncedAt,
		&user.SourceUpdatedAt,
		&user.CreatedAt,
		&user.UpdatedAt,
	)
	if err != nil {
		if IsNotFound(err) {
			return GatewayUser{}, ErrInvalidCredentials
		}
		return GatewayUser{}, err
	}
	if user.Status != "active" || passwordHash == "" {
		return GatewayUser{}, ErrInvalidCredentials
	}
	if err := bcrypt.CompareHashAndPassword([]byte(passwordHash), []byte(input.Password)); err != nil {
		return GatewayUser{}, ErrInvalidCredentials
	}
	user.Roles = decodeStringArray(roles)
	user.AuthProfile = decodeObject(authProfile)
	user.Metadata = decodeObject(metadata)
	_, _ = s.pool.Exec(ctx, `UPDATE gateway_users SET last_login_at=now(), updated_at=now() WHERE id=$1`, user.ID)
	return user, nil
}

func (s *Store) ListRateLimitWindows(ctx context.Context) ([]RateLimitWindow, error) {
	rows, err := s.pool.Query(ctx, `
SELECT scope_type, scope_key, metric, window_start, limit_value::float8, used_value::float8,
       reserved_value::float8, reset_at, updated_at
FROM gateway_rate_limit_counters
WHERE reset_at >= now() - interval '5 minutes'
ORDER BY window_start DESC, scope_type ASC, scope_key ASC, metric ASC`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]RateLimitWindow, 0)
	for rows.Next() {
		var item RateLimitWindow
		if err := rows.Scan(
			&item.ScopeType,
			&item.ScopeKey,
			&item.Metric,
			&item.WindowStart,
			&item.LimitValue,
			&item.UsedValue,
			&item.ReservedValue,
			&item.ResetAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) CreateTask(ctx context.Context, input CreateTaskInput, user *auth.User) (GatewayTask, error) {
	result, err := s.CreateTaskIdempotent(ctx, input, user)
	return result.Task, err
}

func (s *Store) CreateTaskIdempotent(ctx context.Context, input CreateTaskInput, user *auth.User) (CreateTaskResult, error) {
	requestReport := sanitizeJSONForStorageWithReport(input.Request)
	if requestReport.BinaryCount > 0 {
		return CreateTaskResult{}, &taskPayloadBinaryError{
			target: ErrTaskRequestBinaryNotMaterialized,
			code:   "request_binary_not_materialized",
			count:  requestReport.BinaryCount,
		}
	}
	requestBody, _ := json.Marshal(requestReport.Value)
	runMode := normalizeRunMode(input.RunMode, input.Request)
	status := "queued"
	resultBody, _ := json.Marshal(map[string]any(nil))
	billingsBody, _ := json.Marshal([]any(nil))

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return CreateTaskResult{}, err
	}
	defer rollbackTransaction(tx)

	task, err := scanGatewayTask(tx.QueryRow(ctx, `
	INSERT INTO gateway_tasks (
	  external_task_id, kind, run_mode, acceptance_run_id, user_id, gateway_user_id, user_source, gateway_tenant_id, tenant_id, tenant_key,
	  api_key_id, api_key_name, api_key_prefix, user_group_id, user_group_key,
	  model, requested_model, request, async_mode, status, result, billings, conversation_id, new_message_count,
	  idempotency_key_hash, idempotency_request_hash, finished_at
	)
	VALUES (NULLIF($1, ''), $2, $3, NULLIF($4, '')::uuid, $5, NULLIF($6, '')::uuid, COALESCE(NULLIF($7, ''), 'gateway'), NULLIF($8, '')::uuid, NULLIF($9, ''), NULLIF($10, ''), NULLIF($11, ''), NULLIF($12, ''), NULLIF($13, ''), NULLIF($14, '')::uuid, NULLIF($15, ''), $16, $16, $17, $18, $19, $20::jsonb, $21::jsonb, NULLIF($22, '')::uuid, $23, NULLIF($24, ''), NULLIF($25, ''), NULL)
	ON CONFLICT (user_id, idempotency_key_hash) WHERE idempotency_key_hash IS NOT NULL DO NOTHING
	RETURNING `+gatewayTaskColumns,
		strings.TrimSpace(input.ExternalTaskID), input.Kind, runMode, strings.TrimSpace(input.AcceptanceRunID),
		user.ID, user.GatewayUserID, user.Source, user.GatewayTenantID, user.TenantID, user.TenantKey,
		user.APIKeyID, user.APIKeyName, user.APIKeyPrefix, user.UserGroupID, user.UserGroupKey,
		input.Model, requestBody, input.Async, status, resultBody, billingsBody, input.ConversationID,
		input.NewMessageCount, strings.TrimSpace(input.IdempotencyKeyHash), strings.TrimSpace(input.IdempotencyRequestHash),
	))
	replayed := false
	if errors.Is(err, pgx.ErrNoRows) && strings.TrimSpace(input.IdempotencyKeyHash) != "" {
		var existingRequestHash string
		if err := tx.QueryRow(ctx, `
SELECT COALESCE(idempotency_request_hash, '')
FROM gateway_tasks
WHERE user_id = $1 AND idempotency_key_hash = $2
FOR UPDATE`, user.ID, strings.TrimSpace(input.IdempotencyKeyHash)).Scan(&existingRequestHash); err != nil {
			return CreateTaskResult{}, err
		}
		if existingRequestHash != strings.TrimSpace(input.IdempotencyRequestHash) {
			return CreateTaskResult{}, ErrIdempotencyKeyReused
		}
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
SELECT `+gatewayTaskColumns+`
FROM gateway_tasks
WHERE user_id = $1 AND idempotency_key_hash = $2`, user.ID, strings.TrimSpace(input.IdempotencyKeyHash)))
		replayed = true
	}
	if isUniqueViolation(err) && strings.TrimSpace(input.ExternalTaskID) != "" {
		return CreateTaskResult{}, ErrExternalTaskIDReused
	}
	if err != nil {
		return CreateTaskResult{}, err
	}
	if !replayed {
		if err := enforceTaskCapacityTx(ctx, tx, input, runMode); err != nil {
			return CreateTaskResult{}, err
		}
		if err := insertTaskMessageRefs(ctx, tx, task.ID, input.MessageRefs); err != nil {
			return CreateTaskResult{}, err
		}
		events := taskEventsForCreate(task.ID, runMode, status, nil)
		for _, event := range events {
			payload, _ := json.Marshal(sanitizeJSONForStorage(event.Payload))
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_events (task_id, seq, event_type, status, phase, progress, message, payload, simulated)
VALUES ($1::uuid, $2, $3::text, NULLIF($4::text, ''), NULLIF($5::text, ''), $6, NULLIF($7::text, ''), $8::jsonb, $9)`,
				task.ID, event.Seq, event.EventType, event.Status, event.Phase, event.Progress, event.Message, string(payload), event.Simulated,
			); err != nil {
				return CreateTaskResult{}, err
			}
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return CreateTaskResult{}, err
	}
	if replayed {
		task, err = s.GetTask(ctx, task.ID)
		if err != nil {
			return CreateTaskResult{}, err
		}
	}
	return CreateTaskResult{Task: task, Replayed: replayed}, nil
}

func (s *Store) GetTask(ctx context.Context, taskID string) (GatewayTask, error) {
	task, err := scanGatewayTask(s.pool.QueryRow(ctx, `
SELECT `+gatewayTaskColumns+`
	FROM gateway_tasks
	WHERE id=$1`, taskID,
	))
	if err != nil {
		return GatewayTask{}, err
	}
	attempts, err := s.ListTaskAttempts(ctx, task.ID)
	if err != nil {
		return GatewayTask{}, err
	}
	task.Attempts = attempts
	return task, nil
}

func (s *Store) ListTaskStatuses(ctx context.Context, taskIDs []string) (map[string]string, error) {
	statuses := make(map[string]string, len(taskIDs))
	if len(taskIDs) == 0 {
		return statuses, nil
	}
	rows, err := s.pool.Query(ctx, `
SELECT id::text, status
FROM gateway_tasks
WHERE id = ANY($1::uuid[])`, taskIDs)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var taskID string
		var status string
		if err := rows.Scan(&taskID, &status); err != nil {
			return nil, err
		}
		statuses[taskID] = status
	}
	return statuses, rows.Err()
}

type taskScanner interface {
	Scan(dest ...any) error
}

func scanGatewayTask(scanner taskScanner) (GatewayTask, error) {
	var task GatewayTask
	var requestBytes []byte
	var resultBytes []byte
	var billingsBytes []byte
	var usageBytes []byte
	var metricsBytes []byte
	var billingSummaryBytes []byte
	var pricingSnapshotBytes []byte
	var remoteTaskPayloadBytes []byte
	var compatibilitySubmitHeadersBytes []byte
	var compatibilitySubmitBodyBytes []byte
	var routingSnapshotBytes []byte
	var publicErrorBytes []byte
	if err := scanner.Scan(
		&task.ID,
		&task.ExternalTaskID,
		&task.Kind,
		&task.RunMode,
		&task.AcceptanceRunID,
		&task.UserID,
		&task.GatewayUserID,
		&task.UserSource,
		&task.GatewayTenantID,
		&task.TenantID,
		&task.TenantKey,
		&task.APIKeyID,
		&task.APIKeyName,
		&task.APIKeyPrefix,
		&task.UserGroupID,
		&task.UserGroupKey,
		&task.Model,
		&task.ModelType,
		&task.RequestedModel,
		&task.ResolvedModel,
		&task.RequestID,
		&task.ConversationID,
		&task.NewMessageCount,
		&requestBytes,
		&task.AsyncMode,
		&task.RiverJobID,
		&task.Status,
		&task.QueueKey,
		&task.Priority,
		&task.AttemptCount,
		&task.RemoteTaskID,
		&remoteTaskPayloadBytes,
		&resultBytes,
		&billingsBytes,
		&usageBytes,
		&metricsBytes,
		&billingSummaryBytes,
		&task.FinalChargeAmount,
		&task.BillingVersion,
		&task.BillingStatus,
		&task.BillingCurrency,
		&pricingSnapshotBytes,
		&task.RequestFingerprint,
		&task.ReservationAmount,
		&task.ExecutionToken,
		&task.ExecutionLeaseUntil,
		&task.BillingUpdatedAt,
		&task.BillingSettledAt,
		&task.ResponseStartedAt,
		&task.ResponseFinishedAt,
		&task.ResponseDurationMS,
		&task.Error,
		&task.ErrorCode,
		&task.ErrorMessage,
		&publicErrorBytes,
		&task.CompatibilityProtocol,
		&task.CompatibilityPublicID,
		&task.CompatibilitySourceProtocol,
		&task.CompatibilitySubmitHTTPStatus,
		&compatibilitySubmitHeadersBytes,
		&compatibilitySubmitBodyBytes,
		&task.AssignedPoolID,
		&task.AssignedWorkerID,
		&task.RouteProfileKey,
		&task.RoutingPlatformID,
		&task.RoutingPlatformModelID,
		&task.RoutingVersion,
		&task.RoutingReason,
		&routingSnapshotBytes,
		&task.SubmissionState,
		&task.CreatedAt,
		&task.UpdatedAt,
		&task.FinishedAt,
	); err != nil {
		return GatewayTask{}, err
	}
	task.Request = decodeObject(requestBytes)
	task.RemoteTaskPayload = decodeObject(remoteTaskPayloadBytes)
	task.Result = decodeObject(resultBytes)
	task.Billings = decodeArray(billingsBytes)
	task.Usage = decodeObject(usageBytes)
	task.Metrics = decodeObject(metricsBytes)
	task.BillingSummary = decodeObject(billingSummaryBytes)
	task.PricingSnapshot = decodeObject(pricingSnapshotBytes)
	task.CompatibilitySubmitHeaders = decodeObject(compatibilitySubmitHeadersBytes)
	task.CompatibilitySubmitBody = decodeObject(compatibilitySubmitBodyBytes)
	task.RoutingSnapshot = decodeObject(routingSnapshotBytes)
	if len(publicErrorBytes) > 0 {
		var snapshot publicerror.Error
		if json.Unmarshal(publicErrorBytes, &snapshot) == nil && snapshot.Code != "" {
			task.PublicError = &snapshot
		}
	}
	return task, nil
}

func encodePublicErrorSnapshot(code string, message string, status int, retryable bool, requestID string, taskID string) []byte {
	if strings.TrimSpace(code) == "" && strings.TrimSpace(message) == "" {
		return []byte("null")
	}
	snapshot := publicerror.FromFields(code, message, status, retryable)
	snapshot = publicerror.WithIDs(snapshot, requestID, taskID)
	payload, _ := json.Marshal(snapshot)
	return payload
}

func (s *Store) ListTaskEvents(ctx context.Context, taskID string) ([]TaskEvent, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, task_id::text, seq, event_type, COALESCE(status, ''), COALESCE(phase, ''),
       COALESCE(progress, 0)::float8, COALESCE(message, ''), payload, simulated, created_at,
       COALESCE(platform_id::text, '')
FROM gateway_task_events
WHERE task_id = $1::uuid
ORDER BY seq ASC`, taskID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]TaskEvent, 0)
	for rows.Next() {
		var item TaskEvent
		var payload []byte
		if err := rows.Scan(
			&item.ID,
			&item.TaskID,
			&item.Seq,
			&item.EventType,
			&item.Status,
			&item.Phase,
			&item.Progress,
			&item.Message,
			&payload,
			&item.Simulated,
			&item.CreatedAt,
			&item.PlatformID,
		); err != nil {
			return nil, err
		}
		item.Payload = decodeObject(payload)
		items = append(items, item)
	}
	return items, rows.Err()
}

func IsNotFound(err error) bool {
	return err == pgx.ErrNoRows
}

func IsUniqueViolation(err error) bool {
	return isUniqueViolation(err)
}

func IsUndefinedDatabaseObject(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && (pgErr.Code == "42P01" || pgErr.Code == "42703")
}

func isUniqueViolation(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505"
}

type apiKeyScanner interface {
	Scan(dest ...any) error
}

func scanAPIKey(scanner apiKeyScanner) (APIKey, error) {
	var item APIKey
	var scopesBytes []byte
	var rateLimitPolicy []byte
	var quotaPolicy []byte
	if err := scanner.Scan(
		&item.ID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.TenantID,
		&item.TenantKey,
		&item.UserID,
		&item.KeyPrefix,
		&item.Name,
		&scopesBytes,
		&item.UserGroupID,
		&rateLimitPolicy,
		&quotaPolicy,
		&item.Status,
		&item.ExpiresAt,
		&item.LastUsedAt,
		&item.CreatedAt,
		&item.UpdatedAt,
	); err != nil {
		return APIKey{}, err
	}
	item.Scopes = decodeStringArray(scopesBytes)
	item.RateLimitPolicy = decodeObject(rateLimitPolicy)
	item.QuotaPolicy = decodeObject(quotaPolicy)
	return item, nil
}

func scanAPIKeyWithSecret(scanner apiKeyScanner, secret *string) (APIKey, error) {
	var item APIKey
	var scopesBytes []byte
	var rateLimitPolicy []byte
	var quotaPolicy []byte
	if err := scanner.Scan(
		&item.ID,
		&item.GatewayTenantID,
		&item.GatewayUserID,
		&item.TenantID,
		&item.TenantKey,
		&item.UserID,
		&item.KeyPrefix,
		&item.Name,
		&scopesBytes,
		&item.UserGroupID,
		&rateLimitPolicy,
		&quotaPolicy,
		&item.Status,
		&item.ExpiresAt,
		&item.LastUsedAt,
		&item.CreatedAt,
		&item.UpdatedAt,
		secret,
	); err != nil {
		return APIKey{}, err
	}
	item.Scopes = decodeStringArray(scopesBytes)
	item.RateLimitPolicy = decodeObject(rateLimitPolicy)
	item.QuotaPolicy = decodeObject(quotaPolicy)
	if secret != nil {
		item.Secret = *secret
	}
	return item, nil
}

func localGatewayUserID(user *auth.User) string {
	if user == nil {
		return ""
	}
	if user.GatewayUserID != "" {
		return user.GatewayUserID
	}
	if user.Source == "" || user.Source == "gateway" {
		return user.ID
	}
	return ""
}

func generateAPIKeySecret() (string, error) {
	bytes := make([]byte, 32)
	if _, err := rand.Read(bytes); err != nil {
		return "", err
	}
	return "sk-gw-" + base64.RawURLEncoding.EncodeToString(bytes), nil
}

func apiKeyPrefix(secret string) string {
	secret = strings.TrimSpace(secret)
	if !strings.HasPrefix(secret, "sk-gw-") {
		return ""
	}
	if len(secret) <= 18 {
		return secret
	}
	return secret[:18]
}

func normalizeRunMode(input string, request map[string]any) string {
	mode := strings.ToLower(strings.TrimSpace(input))
	if mode == "" && request != nil {
		if raw, ok := request["runMode"].(string); ok {
			mode = strings.ToLower(strings.TrimSpace(raw))
		}
		if raw, ok := request["mode"].(string); ok && mode == "" {
			mode = strings.ToLower(strings.TrimSpace(raw))
		}
		if raw, ok := request["simulation"].(bool); ok && raw {
			mode = "simulation"
		}
		if raw, ok := request["testMode"].(bool); ok && raw {
			mode = "simulation"
		}
	}
	if mode == "test" || mode == "dry-run" || mode == "dry_run" {
		return "simulation"
	}
	if mode == "simulation" {
		return "simulation"
	}
	if mode == "acceptance" {
		return "acceptance"
	}
	if mode == "acceptance_canary" {
		return "acceptance_canary"
	}
	return "production"
}

func simulationResult(kind string, model string) map[string]any {
	switch kind {
	case "chat.completions":
		return map[string]any{
			"id":      "chatcmpl-simulated",
			"object":  "chat.completion",
			"model":   model,
			"choices": []any{map[string]any{"index": 0, "finish_reason": "stop", "message": map[string]any{"role": "assistant", "content": "simulation response"}}},
			"usage":   map[string]any{"prompt_tokens": 12, "completion_tokens": 8, "total_tokens": 20},
		}
	case "images.generations":
		return map[string]any{
			"id":    "img-simulated",
			"model": model,
			"data":  []any{map[string]any{"url": "/static/simulation/image.png", "revised_prompt": "simulation image"}},
		}
	case "videos.generations":
		return map[string]any{
			"id":    "video-simulated",
			"model": model,
			"data":  []any{map[string]any{"url": "/static/simulation/video.mp4", "duration": 5}},
		}
	default:
		return map[string]any{"id": "task-simulated", "model": model, "kind": kind, "ok": true}
	}
}

func simulationBillings(kind string, model string) []any {
	resourceType := "task"
	unit := "item"
	if kind == "chat.completions" {
		resourceType = "text_total"
		unit = "1k_tokens"
	}
	if kind == "images.generations" {
		resourceType = "image"
		unit = "image"
	}
	if kind == "videos.generations" {
		resourceType = "video"
		unit = "second"
	}
	return []any{map[string]any{
		"model":        model,
		"resourceType": resourceType,
		"unit":         unit,
		"quantity":     1,
		"amount":       0,
		"currency":     "resource",
		"simulated":    true,
	}}
}

func taskEventsForCreate(taskID string, runMode string, status string, result map[string]any) []TaskEvent {
	return []TaskEvent{{
		TaskID:    taskID,
		Seq:       1,
		EventType: "task.accepted",
		Status:    "queued",
		Payload:   map[string]any{},
		Simulated: runMode == "simulation",
	}}
}

func decodeObject(bytes []byte) map[string]any {
	if len(bytes) == 0 {
		return nil
	}
	var out map[string]any
	if err := json.Unmarshal(bytes, &out); err != nil {
		return nil
	}
	return out
}

func maskCredentialsPreview(bytes []byte) map[string]any {
	credentials := decodeObject(bytes)
	if len(credentials) == 0 {
		return nil
	}
	out := make(map[string]any, len(credentials))
	for key, value := range credentials {
		out[key] = maskCredentialValue(value)
	}
	return out
}

func maskCredentialValue(value any) any {
	switch typed := value.(type) {
	case string:
		return maskSecret(typed)
	case map[string]any:
		out := make(map[string]any, len(typed))
		for key, nested := range typed {
			out[key] = maskCredentialValue(nested)
		}
		return out
	case []any:
		out := make([]any, 0, len(typed))
		for _, nested := range typed {
			out = append(out, maskCredentialValue(nested))
		}
		return out
	default:
		return value
	}
}

func maskSecret(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return ""
	}
	if len(value) <= 6 {
		return strings.Repeat("*", len(value))
	}
	return value[:3] + strings.Repeat("*", len(value)-6) + value[len(value)-3:]
}

func decodeArray(bytes []byte) []any {
	if len(bytes) == 0 {
		return nil
	}
	var out []any
	if err := json.Unmarshal(bytes, &out); err != nil {
		return nil
	}
	return out
}

func decodeStringArray(bytes []byte) []string {
	if len(bytes) == 0 {
		return nil
	}
	var out []string
	if err := json.Unmarshal(bytes, &out); err == nil {
		return out
	}
	return nil
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func normalizeAccount(value string) string {
	return strings.ToLower(strings.TrimSpace(value))
}

func normalizeKey(value string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	var b strings.Builder
	lastDash := false
	for _, r := range value {
		switch {
		case unicode.IsLetter(r), unicode.IsDigit(r):
			b.WriteRune(r)
			lastDash = false
		case r == '-' || r == '_' || r == '.' || unicode.IsSpace(r):
			if !lastDash && b.Len() > 0 {
				b.WriteByte('-')
				lastDash = true
			}
		}
	}
	out := strings.Trim(b.String(), "-")
	if out == "" {
		return "default"
	}
	return out
}
