Files
sub2api/backend/internal/service/admin_group_duplicate.go
T
Brisbanehuang 20ad5ec506 feat(scheduler): per-group profit control for token account admission
Group pricing (rate multiplier, peak windows, per-user overrides) and
account cost (accounts.rate_multiplier) already live side by side, but
nothing stops the scheduler from handing a request to an account whose
cost multiplier exceeds what the group's pricing can profitably serve.
Add an opt-in per-group profit gate that filters scheduling candidates
by a margin rule, while ordering, scoring, stickiness and breakers keep
working unchanged among qualified accounts.

Admission rule: an account qualifies iff U <= D * (1 - min_margin -
safety_buffer) within a small relative epsilon, where U is
accounts.rate_multiplier (0 is legal; missing/negative/NaN/Inf are
conservatively rejected as invalid) and D is the requester's effective
downstream multiplier (user-group override ?? group default, times the
group peak factor) frozen at the request's pricing instant.

- groups gain profit_control_enabled / profit_min_margin /
  profit_safety_buffer (migration 191); the durable auth-cache
  invalidation trigger additionally watches the profit and pricing
  columns (migration 192) so out-of-band group edits cannot leave
  stale auth snapshots; GetByKeyForAuth explicitly projects the new
  columns and the API-key auth snapshot version is bumped to force a
  refresh of pre-existing snapshots
- request-level pricing instant: token entry points install pricingAt
  into ctx; the profit threshold D and the RecordUsage peak factor
  read the same instant, so one request never changes price mid-flight
  across waits/retries/failover (media and unwired paths keep the
  existing record-time semantics)
- the gate covers token requests on openai, anthropic, gemini, grok
  and antigravity groups: OpenAI-family handlers via
  WithOpenAIRequestPricingContext (responses incl. WS bridge, chat
  completions, messages, embeddings, alpha search), the shared gateway
  via WithGatewayTokenRequestPricing (messages, chat completions,
  responses, gemini model actions); composite groups cannot enable it
  directly; image/video/models/usage/count_tokens stay ungated and an
  explicit image-generation intent suppresses the gate end to end
- post-slot recheck: after a slot is acquired the account is re-read
  via SchedulerSnapshotService.GetAccount (scheduler cache, then DB;
  only when both fail the check fails open with WARN + metric); a
  vetoed account releases its slot and joins the request's exclusion
  set for reselection; sticky bindings are written only after the
  final check passes, and an over-threshold sticky account is skipped,
  not deleted, so it comes back once its rate recovers
- sticky-session cache contract: GatewayCache.GetSessionAccountID now
  returns ErrStickySessionNotFound on a miss (mapped from redis.Nil in
  the repository implementation, mirroring ErrRefreshTokenNotFound) so
  the profit sticky path can distinguish "no binding yet" from a real
  read failure without importing the cache driver in service code
- cross-group re-entry (composite parent -> member group) resolves the
  gate against the member group and clears a stale parent gate instead
  of letting a foreign threshold veto accounts
- per-platform/group activity counters (installs, threshold vetoes,
  invalid-rate vetoes, refresh failures) for observability
- admin UI: profit-control section on the five platforms' group forms
  with percent input, validation and platform-switch reset; group
  create/update/duplicate normalize and validate the config at a
  single choke point
- cmd/profit-preview: offline what-if tool that replays the production
  admission semantics over an exported config/account/override/model
  dump, reports per-model admitted-account counts under the default
  and the worst-case (lowest user override) D, and surfaces probe-sync
  staleness as warnings without affecting admission

Tests: service unit coverage for gate resolution/veto/threshold
epsilon/pricing instant/suppress marker/scheduler filtering and
post-slot recheck (incl. -race on the profit surface), unit-tagged
handler slot-recheck and capability-mapping regressions, sqlmock and
real-PostgreSQL integration regressions for the GetByKeyForAuth
projection and the migration-192 trigger watch list, API contract
update, and frontend specs for the five-platform form helpers.
2026-08-01 22:39:31 +08:00

213 lines
8.2 KiB
Go

package service
import (
"context"
"crypto/sha256"
"errors"
"fmt"
"strconv"
"strings"
"time"
)
const (
maxGroupNameRunes = 100
duplicateGroupInactiveStatus = "inactive"
)
func duplicateGroupOperationID(sourceID int64, actorScope, operationKey string) string {
operationKey = strings.TrimSpace(operationKey)
if operationKey == "" {
return ""
}
actorScope = strings.TrimSpace(actorScope)
if actorScope == "" {
actorScope = "admin:0"
}
payload := "admin.groups.duplicate\x00" + actorScope + "\x00" + strconv.FormatInt(sourceID, 10) + "\x00" + operationKey
digest := sha256.Sum256([]byte(payload))
return fmt.Sprintf("%x", digest)
}
func duplicateGroupName(sourceName string, copyNumber int) string {
if copyNumber < 1 {
copyNumber = 1
}
suffix := " (Copy)"
if copyNumber > 1 {
suffix = fmt.Sprintf(" (Copy %d)", copyNumber)
}
baseRunes := []rune(strings.TrimSpace(sourceName))
maxBaseRunes := maxGroupNameRunes - len([]rune(suffix))
if maxBaseRunes < 0 {
maxBaseRunes = 0
}
if len(baseRunes) > maxBaseRunes {
baseRunes = baseRunes[:maxBaseRunes]
}
return string(baseRunes) + suffix
}
func cloneGroupValuePointer[T any](value *T) *T {
if value == nil {
return nil
}
cloned := *value
return &cloned
}
func cloneGroupModelRouting(value map[string][]int64) map[string][]int64 {
if value == nil {
return nil
}
cloned := make(map[string][]int64, len(value))
for model, accountIDs := range value {
cloned[model] = append([]int64(nil), accountIDs...)
}
return cloned
}
func cloneGroupMessagesDispatchModelConfig(value OpenAIMessagesDispatchModelConfig) OpenAIMessagesDispatchModelConfig {
cloned := value
if value.ExactModelMappings != nil {
cloned.ExactModelMappings = make(map[string]string, len(value.ExactModelMappings))
for requestedModel, mappedModel := range value.ExactModelMappings {
cloned.ExactModelMappings[requestedModel] = mappedModel
}
}
return cloned
}
func cloneGroupForDuplicate(source *Group, operationID string) *Group {
return &Group{
Name: duplicateGroupName(source.Name, 1),
Description: source.Description,
Platform: source.Platform,
RateMultiplier: source.RateMultiplier,
PeakRateEnabled: source.PeakRateEnabled,
PeakStart: source.PeakStart,
PeakEnd: source.PeakEnd,
PeakRateMultiplier: source.PeakRateMultiplier,
ProfitControlEnabled: source.ProfitControlEnabled,
ProfitMinMargin: source.ProfitMinMargin,
ProfitSafetyBuffer: source.ProfitSafetyBuffer,
IsExclusive: source.IsExclusive,
Status: duplicateGroupInactiveStatus,
DuplicateOperationID: operationID,
SubscriptionType: source.SubscriptionType,
DailyLimitUSD: cloneGroupValuePointer(source.DailyLimitUSD),
WeeklyLimitUSD: cloneGroupValuePointer(source.WeeklyLimitUSD),
MonthlyLimitUSD: cloneGroupValuePointer(source.MonthlyLimitUSD),
DefaultValidityDays: source.DefaultValidityDays,
AllowImageGeneration: source.AllowImageGeneration,
AllowBatchImageGeneration: source.AllowBatchImageGeneration,
ImageRateIndependent: source.ImageRateIndependent,
ImageRateMultiplier: source.ImageRateMultiplier,
ImagePrice1K: cloneGroupValuePointer(source.ImagePrice1K),
ImagePrice2K: cloneGroupValuePointer(source.ImagePrice2K),
ImagePrice4K: cloneGroupValuePointer(source.ImagePrice4K),
BatchImageDiscountMultiplier: source.BatchImageDiscountMultiplier,
BatchImageHoldMultiplier: source.BatchImageHoldMultiplier,
VideoRateIndependent: source.VideoRateIndependent,
VideoRateMultiplier: source.VideoRateMultiplier,
VideoPrice480P: cloneGroupValuePointer(source.VideoPrice480P),
VideoPrice720P: cloneGroupValuePointer(source.VideoPrice720P),
VideoPrice1080P: cloneGroupValuePointer(source.VideoPrice1080P),
WebSearchPricePerCall: cloneGroupValuePointer(source.WebSearchPricePerCall),
ClaudeCodeOnly: source.ClaudeCodeOnly,
FallbackGroupID: cloneGroupValuePointer(source.FallbackGroupID),
FallbackGroupIDOnInvalidRequest: cloneGroupValuePointer(source.FallbackGroupIDOnInvalidRequest),
ModelRouting: cloneGroupModelRouting(source.ModelRouting),
ModelRoutingEnabled: source.ModelRoutingEnabled,
MCPXMLInject: source.MCPXMLInject,
SupportedModelScopes: append([]string(nil), source.SupportedModelScopes...),
SortOrder: source.SortOrder,
AllowMessagesDispatch: source.AllowMessagesDispatch,
AllowLive: source.AllowLive,
RequireOAuthOnly: source.RequireOAuthOnly,
RequirePrivacySet: source.RequirePrivacySet,
DefaultMappedModel: source.DefaultMappedModel,
MessagesDispatchModelConfig: cloneGroupMessagesDispatchModelConfig(source.MessagesDispatchModelConfig),
ModelsListConfig: GroupModelsListConfig{
Enabled: source.ModelsListConfig.Enabled,
Models: append([]string(nil), source.ModelsListConfig.Models...),
},
RPMLimit: source.RPMLimit,
MaxReasoningEffort: source.MaxReasoningEffort,
ReasoningEffortMappings: append([]ReasoningEffortMapping(nil), source.ReasoningEffortMappings...),
}
}
// RecoverDuplicateGroup performs a read-only lookup for a copy that was already
// committed for the same actor, source group, and idempotency key.
func (s *adminServiceImpl) RecoverDuplicateGroup(ctx context.Context, id int64, actorScope, operationKey string) (*Group, error) {
operationID := duplicateGroupOperationID(id, actorScope, operationKey)
if operationID == "" {
return nil, nil
}
if s.groupDuplicateRepo == nil {
return nil, errors.New("group duplicate repository is not configured")
}
group, err := s.groupDuplicateRepo.FindByDuplicateOperationID(ctx, operationID)
if err != nil {
return nil, fmt.Errorf("find duplicate group operation: %w", err)
}
if group == nil {
return nil, nil
}
hydrated, err := s.groupRepo.GetByID(ctx, group.ID)
if err != nil {
return nil, fmt.Errorf("load recovered duplicate group: %w", err)
}
return hydrated, nil
}
// DuplicateGroup creates an inactive copy of a group's configuration and exact
// account priorities. The repository commits the group, bindings, and outbox
// event atomically so a failed binding never leaves an orphan group.
func (s *adminServiceImpl) DuplicateGroup(ctx context.Context, id int64, actorScope, operationKey string) (*Group, error) {
existing, err := s.RecoverDuplicateGroup(ctx, id, actorScope, operationKey)
if err != nil {
return nil, err
}
if existing != nil {
return existing, nil
}
source, err := s.groupRepo.GetByID(ctx, id)
if err != nil {
return nil, err
}
if s.groupDuplicateRepo == nil {
return nil, errors.New("group duplicate repository is not configured")
}
duplicate := cloneGroupForDuplicate(source, duplicateGroupOperationID(id, actorScope, operationKey))
sanitizeGroupReasoningEffortPolicy(duplicate)
for copyNumber := 1; ; copyNumber++ {
duplicate.Name = duplicateGroupName(source.Name, copyNumber)
duplicate.ID = 0
duplicate.CreatedAt = time.Time{}
duplicate.UpdatedAt = time.Time{}
if err := s.groupDuplicateRepo.CreateFromSource(ctx, duplicate, source.ID); err == nil {
hydrated, loadErr := s.groupRepo.GetByID(ctx, duplicate.ID)
if loadErr != nil {
return nil, fmt.Errorf("load duplicate group: %w", loadErr)
}
return hydrated, nil
} else if !errors.Is(err, ErrGroupExists) {
return nil, fmt.Errorf("create duplicate group: %w", err)
}
// A unique conflict can be either the generated name or the operation ID.
// Recover first; if no operation row exists, advance to the next name.
recovered, recoverErr := s.RecoverDuplicateGroup(ctx, id, actorScope, operationKey)
if recoverErr != nil {
return nil, recoverErr
}
if recovered != nil {
return recovered, nil
}
}
}