From da2bf242250ade85864f4626b60c26a7cca88c22 Mon Sep 17 00:00:00 2001 From: alfadb Date: Thu, 13 Aug 2026 23:41:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(opencode):=20=E7=94=A8=E9=87=8F=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E6=94=AF=E6=8C=81=E5=90=8C=20Key=20=E7=BB=84=E5=85=B1?= =?UTF-8?q?=E4=BA=AB=E3=80=81=E6=B4=BB=E5=8A=A8=E9=98=B2=E6=8A=96=E4=B8=8E?= =?UTF-8?q?=E8=B6=85=E7=AA=97=E5=BC=BA=E5=88=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与 Ollama Cloud usage 对齐：OpenCode Go 作为同 Key 聚合订阅， 多个 openai/apikey/base_url=opencode.ai/zen/go/v1 账号共享同一份 用量状态与刷新节奏： - 组共享：组指纹 sha256("opencode.ai\0"+api_key)；auto_refresh 开关 与 snapshot 按组写（事务 + FOR NO KEY UPDATE + 锚点 CAS + 纯合并， 写 snapshot 不会抹掉开关）；列表/详情经 ResolveAccounts 组内共享； 账号换 Key → IdentityChanged → 组级关闭 auto_refresh 并清快照 - 活动防抖 + 超窗强刷：settings 新增 debounce_minutes（默认 1， 1-60，必须小于 interval_minutes）；模型请求活动经网关 5 处调用点 驱动刷新，due = min(lastUsed+debounce, fetchedAt+maxWait)， 成功路径有 5 分钟最小抓取间隔，失败路径退避优先 - 手动刷新限频 10s→30s 且按组；singleflight/RunDue 按组去重； ListDue 在 SQL 内按组算 due（CTE 分组取组内 MAX(last_used_at)） 验证：go build + 五包回归 OK；新增组共享/三态 due/组单飞/ IdentityChanged/debounce 校验单测全绿；vitest 11 例全绿； vue-tsc 零错误。 --- .../internal/handler/admin/account_handler.go | 22 +- .../admin/account_opencode_go_usage_test.go | 38 +- .../account_repo_opencode_go_usage.go | 428 ++++++++++++++-- .../service/gateway_anthropic_passthrough.go | 1 + backend/internal/service/gateway_forward.go | 3 +- .../service/gateway_upstream_response.go | 3 +- .../openai_account_runtime_block_fastpath.go | 1 + .../openai_upstream_transport_error.go | 3 +- backend/internal/service/opencode_go_usage.go | 342 ++++++++++++- .../service/opencode_go_usage_test.go | 458 ++++++++++++++++-- .../src/i18n/locales/en/admin/settings.ts | 6 +- .../src/i18n/locales/zh/admin/settings.ts | 6 +- frontend/src/types/index.ts | 3 + frontend/src/views/admin/SettingsView.vue | 21 +- 14 files changed, 1230 insertions(+), 105 deletions(-) diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index 0e6e1029c8c7..23dbbd396793 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -543,14 +543,22 @@ func (h *AccountHandler) List(c *gin.Context) { response.ErrorFrom(c, err) return } - if h.ollamaCloudUsage != nil && len(accounts) > 0 { + if len(accounts) > 0 { accountPointers := make([]*service.Account, len(accounts)) for index := range accounts { accountPointers[index] = &accounts[index] } - if err := h.ollamaCloudUsage.ResolveAccounts(c.Request.Context(), accountPointers); err != nil { - response.ErrorFrom(c, err) - return + if h.ollamaCloudUsage != nil { + if err := h.ollamaCloudUsage.ResolveAccounts(c.Request.Context(), accountPointers); err != nil { + response.ErrorFrom(c, err) + return + } + } + if h.opencodeGoUsage != nil { + if err := h.opencodeGoUsage.ResolveOpenCodeGoUsageAccounts(c.Request.Context(), accountPointers); err != nil { + response.ErrorFrom(c, err) + return + } } } @@ -776,6 +784,12 @@ func (h *AccountHandler) GetByID(c *gin.Context) { return } } + if h.opencodeGoUsage != nil { + if err := h.opencodeGoUsage.ResolveOpenCodeGoUsageAccounts(c.Request.Context(), []*service.Account{account}); err != nil { + response.ErrorFrom(c, err) + return + } + } response.Success(c, h.buildAccountResponseWithRuntime(c.Request.Context(), account)) } diff --git a/backend/internal/handler/admin/account_opencode_go_usage_test.go b/backend/internal/handler/admin/account_opencode_go_usage_test.go index e1ad0015feef..b763367d27d3 100644 --- a/backend/internal/handler/admin/account_opencode_go_usage_test.go +++ b/backend/internal/handler/admin/account_opencode_go_usage_test.go @@ -32,13 +32,49 @@ func (r *openCodeGoUsageHandlerTestRepo) GetByID(_ context.Context, id int64) (* return nil, service.ErrAccountNotFound } +func (r *openCodeGoUsageHandlerTestRepo) ListOpenCodeGoUsageGroupAccounts(_ context.Context, anchors []*service.Account) ([]service.Account, error) { + wanted := make(map[string]struct{}, len(anchors)) + for _, anchor := range anchors { + if apiKey, ok := openCodeGoUsageHandlerTestAPIKey(anchor); ok { + wanted[apiKey] = struct{}{} + } + } + result := make([]service.Account, 0, len(r.accounts)+1) + if r.account != nil { + if apiKey, ok := openCodeGoUsageHandlerTestAPIKey(r.account); ok { + if _, match := wanted[apiKey]; match { + result = append(result, *r.account) + } + } + } + for _, account := range r.accounts { + if apiKey, ok := openCodeGoUsageHandlerTestAPIKey(account); ok { + if _, match := wanted[apiKey]; match { + result = append(result, *account) + } + } + } + return result, nil +} + +func openCodeGoUsageHandlerTestAPIKey(account *service.Account) (string, bool) { + if account == nil || account.Credentials == nil { + return "", false + } + apiKey, ok := account.Credentials["api_key"].(string) + return apiKey, ok && apiKey != "" +} + func (r *openCodeGoUsageHandlerTestRepo) SetOpenCodeGoUsageAutoRefresh(context.Context, *service.Account, bool) error { return nil } func (r *openCodeGoUsageHandlerTestRepo) UpdateOpenCodeGoUsageSnapshot(context.Context, *service.Account, *service.OpenCodeGoUsageSnapshot) error { return nil } -func (r *openCodeGoUsageHandlerTestRepo) ListDueOpenCodeGoUsageAccounts(context.Context, time.Time, int) ([]service.Account, error) { +func (r *openCodeGoUsageHandlerTestRepo) DisableOpenCodeGoUsageAutoRefresh(context.Context, *service.Account) error { + return nil +} +func (r *openCodeGoUsageHandlerTestRepo) ListDueOpenCodeGoUsageAccounts(context.Context, time.Time, time.Duration, time.Duration, int) ([]service.Account, error) { return nil, nil } diff --git a/backend/internal/repository/account_repo_opencode_go_usage.go b/backend/internal/repository/account_repo_opencode_go_usage.go index ef645301ca55..63de8e3ed6ca 100644 --- a/backend/internal/repository/account_repo_opencode_go_usage.go +++ b/backend/internal/repository/account_repo_opencode_go_usage.go @@ -8,6 +8,7 @@ import ( dbent "github.com/Wei-Shaw/sub2api/ent" "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/lib/pq" ) const ( @@ -22,7 +23,69 @@ const ( ` ) -// SetOpenCodeGoUsageAutoRefresh persists the per-account auto-refresh switch. +// ListOpenCodeGoUsageGroupAccounts resolves every sibling for all supplied +// identities with one ID query and one batch hydration. API keys are query +// parameters only; no derived shared key is persisted. +func (r *accountRepository) ListOpenCodeGoUsageGroupAccounts(ctx context.Context, accounts []*service.Account) ([]service.Account, error) { + if r == nil || r.sql == nil { + return nil, service.ErrOpenCodeGoUsageUnavailable + } + keys := make([]string, 0, len(accounts)) + seen := make(map[string]struct{}, len(accounts)) + for _, account := range accounts { + if !service.IsOpenCodeGoUsageAccount(account) || account.Credentials == nil { + continue + } + apiKey, ok := account.Credentials["api_key"].(string) + if !ok || apiKey == "" { + continue + } + if _, duplicate := seen[apiKey]; duplicate { + continue + } + seen[apiKey] = struct{}{} + keys = append(keys, apiKey) + } + if len(keys) == 0 { + return []service.Account{}, nil + } + rows, err := r.sql.QueryContext(ctx, ` + SELECT id + FROM accounts + WHERE deleted_at IS NULL + AND `+opencodeGoUsageEligibleSQL+` + AND credentials ->> 'api_key' = ANY($1) + ORDER BY id + `, pq.Array(keys)) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + ids := make([]int64, 0, len(keys)) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + hydrated, err := r.GetByIDs(ctx, ids) + if err != nil { + return nil, err + } + result := make([]service.Account, 0, len(hydrated)) + for _, account := range hydrated { + if account != nil { + result = append(result, *account) + } + } + return result, nil +} + +// SetOpenCodeGoUsageAutoRefresh persists the group-scoped auto-refresh switch. func (r *accountRepository) SetOpenCodeGoUsageAutoRefresh(ctx context.Context, account *service.Account, enabled bool) error { if account == nil { return service.ErrAccountNilInput @@ -30,12 +93,12 @@ func (r *accountRepository) SetOpenCodeGoUsageAutoRefresh(ctx context.Context, a if r == nil || r.client == nil || !service.IsOpenCodeGoUsageAccount(account) { return service.ErrOpenCodeGoUsageUnavailable } - return r.updateOpenCodeGoUsageExtra(ctx, account, map[string]any{ + return r.updateOpenCodeGoUsageGroup(ctx, account, map[string]any{ service.OpenCodeGoUsageAutoRefreshExtraKey: enabled, - }) + }, nil, true) } -// UpdateOpenCodeGoUsageSnapshot persists the per-account usage snapshot. +// UpdateOpenCodeGoUsageSnapshot persists the group-scoped usage snapshot. func (r *accountRepository) UpdateOpenCodeGoUsageSnapshot(ctx context.Context, account *service.Account, snapshot *service.OpenCodeGoUsageSnapshot) error { if account == nil || snapshot == nil { return service.ErrAccountNilInput @@ -43,27 +106,135 @@ func (r *accountRepository) UpdateOpenCodeGoUsageSnapshot(ctx context.Context, a if r == nil || r.client == nil || !service.IsOpenCodeGoUsageAccount(account) { return service.ErrOpenCodeGoUsageUnavailable } - return r.updateOpenCodeGoUsageExtra(ctx, account, map[string]any{ - service.OpenCodeGoUsageSnapshotExtraKey: snapshot, - }) + payload := openCodeGoUsageManagedPayload(account) + payload[service.OpenCodeGoUsageSnapshotExtraKey] = snapshot + return r.updateOpenCodeGoUsageGroup(ctx, account, payload, nil, true) } -// updateOpenCodeGoUsageExtra atomically merges managed extra keys onto the -// account row. Snapshots are written per account (no api_key group CAS). -func (r *accountRepository) updateOpenCodeGoUsageExtra(ctx context.Context, account *service.Account, payload map[string]any) error { - encoded, err := json.Marshal(payload) - if err != nil { - return err +// DisableOpenCodeGoUsageAutoRefresh is group-scoped and retains the loaded +// identity CAS. It cannot disable a new group after the account changes key. +func (r *accountRepository) DisableOpenCodeGoUsageAutoRefresh(ctx context.Context, account *service.Account) error { + if account == nil { + return service.ErrAccountNilInput + } + if r == nil || r.client == nil || !service.IsOpenCodeGoUsageAccount(account) { + return service.ErrOpenCodeGoUsageUnavailable + } + payload := openCodeGoUsageManagedPayload(account) + payload[service.OpenCodeGoUsageAutoRefreshExtraKey] = false + delete(payload, service.OpenCodeGoUsageSnapshotExtraKey) + return r.updateOpenCodeGoUsageGroup(ctx, account, payload, + []string{service.OpenCodeGoUsageSnapshotExtraKey}, true) +} + +func openCodeGoUsageManagedPayload(account *service.Account) map[string]any { + payload := make(map[string]any, 2) + if account == nil || account.Extra == nil { + return payload + } + for _, key := range []string{ + service.OpenCodeGoUsageAutoRefreshExtraKey, + service.OpenCodeGoUsageSnapshotExtraKey, + } { + if value, ok := account.Extra[key]; ok { + payload[key] = value + } + } + return payload +} + +type lockedOpenCodeGoUsageMember struct { + id int64 + anchorMatches bool + autoJSON string + snapshotJSON string +} + +// updateOpenCodeGoUsageGroup locks every member of the exact api_key group and +// merges the payload onto each member's extra. The merge is pure +// (COALESCE(extra,'{}') || payload) unless deleteKeys is non-empty, in which +// case those keys are removed first — used only by the group-level disable path +// to clear the stale snapshot. Managed keys are never removed on ordinary +// writes, so a snapshot write can never wipe the auto-refresh switch or vice versa. +func (r *accountRepository) updateOpenCodeGoUsageGroup( + ctx context.Context, + account *service.Account, + payload map[string]any, + deleteKeys []string, + requireExpectedState bool, +) error { + if account == nil { + return service.ErrAccountNilInput + } + if r == nil || r.client == nil || !service.IsOpenCodeGoUsageAccount(account) { + return service.ErrOpenCodeGoUsageUnavailable + } + apiKey, ok := account.Credentials["api_key"].(string) + if !ok || apiKey == "" { + return service.ErrOpenCodeGoUsageAccountInvalid } apply := func(txCtx context.Context, client *dbent.Client) error { + matchesProxy, err := lockAndMatchProbeProxyIdentity(txCtx, client, account) + if err != nil { + return err + } + if !matchesProxy { + return service.ErrOpenCodeGoUsageIdentityChanged + } + members, err := lockOpenCodeGoUsageGroup(txCtx, client, account, apiKey) + if err != nil { + return err + } + anchorMatches := false + for _, member := range members { + anchorMatches = anchorMatches || member.anchorMatches + } + if !anchorMatches { + return service.ErrOpenCodeGoUsageIdentityChanged + } + if requireExpectedState { + expectedAuto, err := canonicalAccountExtraJSON(account, service.OpenCodeGoUsageAutoRefreshExtraKey) + if err != nil { + return err + } + expectedSnapshot, err := canonicalAccountExtraJSON(account, service.OpenCodeGoUsageSnapshotExtraKey) + if err != nil { + return err + } + stateMatches := false + for _, member := range members { + if canonicalJSON(member.autoJSON) == expectedAuto && + canonicalJSON(member.snapshotJSON) == expectedSnapshot { + stateMatches = true + break + } + } + if !stateMatches { + return service.ErrOpenCodeGoUsageIdentityChanged + } + } + encoded, err := json.Marshal(payload) + if err != nil { + return err + } + memberIDs := make([]int64, len(members)) + for index := range members { + memberIDs[index] = members[index].id + } + mergeExpr := "COALESCE(extra, '{}'::jsonb)" + for _, key := range deleteKeys { + mergeExpr += " - " + pq.QuoteLiteral(key) + } + mergeExpr += " || $1::jsonb" result, err := client.ExecContext(txCtx, ` UPDATE accounts - SET extra = COALESCE(extra, '{}'::jsonb) || $1::jsonb, + SET extra = `+mergeExpr+`, updated_at = NOW() WHERE deleted_at IS NULL AND `+opencodeGoUsageEligibleSQL+` - AND id = $2 - `, string(encoded), account.ID) + AND credentials ->> 'api_key' = $2 + AND id = ANY($3) + `, string(encoded), apiKey, pq.Array(memberIDs)) if err != nil { return err } @@ -71,7 +242,7 @@ func (r *accountRepository) updateOpenCodeGoUsageExtra(ctx context.Context, acco if err != nil { return err } - if affected != 1 { + if affected != int64(len(members)) { return service.ErrOpenCodeGoUsageIdentityChanged } return nil @@ -94,57 +265,226 @@ func (r *accountRepository) updateOpenCodeGoUsageExtra(ctx context.Context, acco return tx.Commit() } -// ListDueOpenCodeGoUsageAccounts returns at most limit eligible accounts whose -// auto-refresh is enabled and whose snapshot is missing or due (next_refresh_at -// at or before now). Invalid/missing next_refresh_at values fail open to due. -func (r *accountRepository) ListDueOpenCodeGoUsageAccounts(ctx context.Context, now time.Time, limit int) ([]service.Account, error) { +func lockOpenCodeGoUsageGroup( + ctx context.Context, + client *dbent.Client, + account *service.Account, + apiKey string, +) ([]lockedOpenCodeGoUsageMember, error) { + credentials, err := json.Marshal(normalizeJSONMap(account.Credentials)) + if err != nil { + return nil, err + } + var proxyID any + if account.ProxyID != nil { + proxyID = *account.ProxyID + } + rows, err := client.QueryContext(ctx, ` + SELECT + id, + id = $2 + AND platform = $3 + AND type = $4 + AND credentials = $5::jsonb + AND proxy_id IS NOT DISTINCT FROM $6, + COALESCE((extra -> 'opencode_go_usage_auto_refresh')::text, 'null'), + COALESCE((extra -> 'opencode_go_usage_snapshot')::text, 'null') + FROM accounts + WHERE deleted_at IS NULL + AND `+opencodeGoUsageEligibleSQL+` + AND credentials ->> 'api_key' = $1 + ORDER BY id + FOR NO KEY UPDATE + `, apiKey, account.ID, account.Platform, account.Type, string(credentials), proxyID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + members := make([]lockedOpenCodeGoUsageMember, 0, 1) + for rows.Next() { + var member lockedOpenCodeGoUsageMember + if err := rows.Scan(&member.id, &member.anchorMatches, &member.autoJSON, &member.snapshotJSON); err != nil { + return nil, err + } + members = append(members, member) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(members) == 0 { + return nil, service.ErrOpenCodeGoUsageIdentityChanged + } + return members, nil +} + +// ListDueOpenCodeGoUsageAccounts returns at most one truly-due activity-driven +// candidate per exact API key. Due timing (debounce, max-wait, failure backoff) +// is evaluated in SQL before LIMIT so non-due active groups cannot starve due ones. +// Account.LastUsedAt is stamped with the group MAX(last_used_at) for a service +// pure-function recheck against races between list and refresh. +// +// Rules mirror service.openCodeGoUsageAutoRefreshDueAt (keep both in sync): +// - missing/invalid snapshot or times → fail-open first due +// - success: activity after fetched_at; +// due_at = GREATEST(LEAST(last_used+debounce, fetched+maxWait), fetched+minFetchInterval) +// - failed/unauthorized: activity after last_attempt; activity_due = LEAST(...); +// final due_at is not earlier than a valid next_refresh_at (invalid/missing fail-open) +func (r *accountRepository) ListDueOpenCodeGoUsageAccounts( + ctx context.Context, + now time.Time, + debounce, maxWait time.Duration, + limit int, +) ([]service.Account, error) { if limit <= 0 { return []service.Account{}, nil } if r == nil || r.sql == nil { return nil, errors.New("account repository SQL executor not configured") } - nextRefreshExpr := "extra -> 'opencode_go_usage_snapshot' #>> '{next_refresh_at}'" + if debounce <= 0 { + debounce = time.Minute + } + if maxWait <= 0 { + maxWait = 15 * time.Minute + } + debounceSeconds := debounce.Seconds() + maxWaitSeconds := maxWait.Seconds() + minFetchIntervalSeconds := service.OpenCodeGoUsageMinFetchInterval.Seconds() rows, err := r.sql.QueryContext(ctx, ` - SELECT id - FROM accounts - WHERE deleted_at IS NULL - AND status = 'active' - AND `+opencodeGoUsageEligibleSQL+` - AND extra @> '{"opencode_go_usage_auto_refresh": true}'::jsonb - AND ( - extra -> 'opencode_go_usage_snapshot' IS NULL - OR extra -> 'opencode_go_usage_snapshot' = 'null'::jsonb - OR `+ollamaCloudUsageParseRFC3339SQL(nextRefreshExpr)+` IS NULL - OR `+ollamaCloudUsageParseRFC3339SQL(nextRefreshExpr)+`::timestamptz <= $1 - ) - ORDER BY id - LIMIT $2 - `, now.UTC(), limit) + WITH eligible AS ( + SELECT id, + credentials ->> 'api_key' AS api_key, + last_used_at, + extra -> 'opencode_go_usage_snapshot' AS snapshot + FROM accounts + WHERE deleted_at IS NULL + AND status = 'active' + AND `+opencodeGoUsageEligibleSQL+` + AND extra @> '{"opencode_go_usage_auto_refresh": true}'::jsonb + ), group_activity AS ( + SELECT credentials ->> 'api_key' AS api_key, + MAX(last_used_at) AS group_last_used_at + FROM accounts + WHERE deleted_at IS NULL + AND `+opencodeGoUsageEligibleSQL+` + AND jsonb_typeof(credentials -> 'api_key') = 'string' + GROUP BY credentials ->> 'api_key' + ), joined AS ( + SELECT e.id, e.api_key, e.snapshot, g.group_last_used_at, + e.snapshot #>> '{status}' AS status, + e.snapshot #>> '{fetched_at}' AS fetched_at, + e.snapshot #>> '{last_attempt_at}' AS last_attempt_at, + e.snapshot #>> '{next_refresh_at}' AS next_refresh_at + FROM eligible e + JOIN group_activity g ON g.api_key = e.api_key + ), parsed AS MATERIALIZED ( + SELECT id, api_key, snapshot, group_last_used_at, status, + `+ollamaCloudUsageParseRFC3339SQL("fetched_at")+` AS parsed_fetched_at, + `+ollamaCloudUsageParseRFC3339SQL("last_attempt_at")+` AS parsed_last_attempt_at, + `+ollamaCloudUsageParseRFC3339SQL("next_refresh_at")+` AS parsed_next_refresh_at + FROM joined + ), timed AS ( + SELECT *, + CASE + WHEN status = 'ok' + AND parsed_fetched_at IS NOT NULL + AND group_last_used_at IS NOT NULL + AND group_last_used_at > parsed_fetched_at::timestamptz + THEN GREATEST( + LEAST( + group_last_used_at + make_interval(secs => $2::double precision), + parsed_fetched_at::timestamptz + make_interval(secs => $3::double precision) + ), + parsed_fetched_at::timestamptz + make_interval(secs => $5::double precision) + ) + WHEN status IN ('failed', 'unauthorized') + AND parsed_last_attempt_at IS NOT NULL + AND group_last_used_at IS NOT NULL + AND group_last_used_at > parsed_last_attempt_at::timestamptz + THEN GREATEST( + LEAST( + group_last_used_at + make_interval(secs => $2::double precision), + parsed_last_attempt_at::timestamptz + make_interval(secs => $3::double precision) + ), + COALESCE(parsed_next_refresh_at::timestamptz, '-infinity'::timestamptz) + ) + ELSE NULL + END AS activity_due_at + FROM parsed + ), candidates AS ( + SELECT *, + CASE + WHEN snapshot IS NULL OR snapshot = 'null'::jsonb OR status IS NULL + OR status NOT IN ('ok', 'failed', 'unauthorized') THEN 0 + WHEN status = 'ok' AND parsed_fetched_at IS NULL THEN 0 + WHEN status IN ('failed', 'unauthorized') AND parsed_last_attempt_at IS NULL THEN 0 + WHEN activity_due_at IS NOT NULL AND $1 >= activity_due_at THEN 1 + ELSE NULL + END AS due_class, + activity_due_at AS due_at + FROM timed + ), ranked AS ( + SELECT id, api_key, group_last_used_at, due_class, due_at, + row_number() OVER ( + PARTITION BY api_key + ORDER BY due_class, + due_at NULLS FIRST, + id + ) AS group_rank + FROM candidates + WHERE due_class IS NOT NULL + ) + SELECT id, group_last_used_at + FROM ranked + WHERE group_rank = 1 + ORDER BY due_class, due_at NULLS FIRST, id + LIMIT $4 + `, now.UTC(), debounceSeconds, maxWaitSeconds, limit, minFetchIntervalSeconds) if err != nil { return nil, err } defer func() { _ = rows.Close() }() + type dueRow struct { + id int64 + groupLastUsed *time.Time + } + rowsOut := make([]dueRow, 0, limit) ids := make([]int64, 0, limit) for rows.Next() { - var id int64 - if err := rows.Scan(&id); err != nil { + var row dueRow + if err := rows.Scan(&row.id, &row.groupLastUsed); err != nil { return nil, err } - ids = append(ids, id) + rowsOut = append(rowsOut, row) + ids = append(ids, row.id) } if err := rows.Err(); err != nil { return nil, err } - hydrated, err := r.GetByIDs(ctx, ids) + accounts, err := r.GetByIDs(ctx, ids) if err != nil { return nil, err } - result := make([]service.Account, 0, len(hydrated)) - for _, account := range hydrated { + byID := make(map[int64]*service.Account, len(accounts)) + for _, account := range accounts { if account != nil { - result = append(result, *account) + byID[account.ID] = account + } + } + result := make([]service.Account, 0, len(rowsOut)) + for _, row := range rowsOut { + account := byID[row.id] + if account == nil { + continue + } + // Stamp group MAX(last_used_at) for service due evaluation. + if row.groupLastUsed != nil { + ts := row.groupLastUsed.UTC() + account.LastUsedAt = &ts + } else { + account.LastUsedAt = nil } + result = append(result, *account) } return result, nil } diff --git a/backend/internal/service/gateway_anthropic_passthrough.go b/backend/internal/service/gateway_anthropic_passthrough.go index c18d5bf047cd..1f3e08f3d340 100644 --- a/backend/internal/service/gateway_anthropic_passthrough.go +++ b/backend/internal/service/gateway_anthropic_passthrough.go @@ -116,6 +116,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput( } if !errors.Is(err, context.Canceled) { scheduleOllamaCloudUsageActivity(s.deferredService, account) + scheduleOpenCodeGoUsageActivity(s.deferredService, account) } safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") diff --git a/backend/internal/service/gateway_forward.go b/backend/internal/service/gateway_forward.go index d10cb1ea3be5..f7204c9f6f58 100644 --- a/backend/internal/service/gateway_forward.go +++ b/backend/internal/service/gateway_forward.go @@ -381,9 +381,10 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A if resp != nil && resp.Body != nil { _ = resp.Body.Close() } - // Transport attempt left local validation; count Ollama Cloud activity. + // Transport attempt left local validation; count Ollama Cloud / OpenCode Go activity. if !errors.Is(err, context.Canceled) { scheduleOllamaCloudUsageActivity(s.deferredService, account) + scheduleOpenCodeGoUsageActivity(s.deferredService, account) } // Ensure the client receives an error response (handlers assume Forward writes on non-failover errors). safeErr := sanitizeUpstreamErrorMessage(err.Error()) diff --git a/backend/internal/service/gateway_upstream_response.go b/backend/internal/service/gateway_upstream_response.go index 5a9fedf96433..caeb1835ca7b 100644 --- a/backend/internal/service/gateway_upstream_response.go +++ b/backend/internal/service/gateway_upstream_response.go @@ -356,8 +356,9 @@ func (s *GatewayService) readUpstreamErrorBody(resp *http.Response) ([]byte, err } func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, requestedModel ...string) (*ForwardResult, error) { - // Upstream returned a non-success HTTP status; count Ollama Cloud activity. + // Upstream returned a non-success HTTP status; count Ollama Cloud / OpenCode Go activity. scheduleOllamaCloudUsageActivity(s.deferredService, account) + scheduleOpenCodeGoUsageActivity(s.deferredService, account) body, readErr := s.readUpstreamErrorBody(resp) if readErr != nil { // 读取失败时 body 可能被截断，错误分类会基于不完整数据；记录日志以便排查， diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index ace8366a0eb0..73d06900c5bc 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -54,6 +54,7 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont // Any non-2xx upstream HTTP response means the model request was actually sent. if s != nil { scheduleOllamaCloudUsageActivity(s.deferredService, account) + scheduleOpenCodeGoUsageActivity(s.deferredService, account) } stateCtx, cancel := openAIAccountStateContext(ctx) defer cancel() diff --git a/backend/internal/service/openai_upstream_transport_error.go b/backend/internal/service/openai_upstream_transport_error.go index 4093ca0b64cd..383ff924e0be 100644 --- a/backend/internal/service/openai_upstream_transport_error.go +++ b/backend/internal/service/openai_upstream_transport_error.go @@ -124,9 +124,10 @@ func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Co return err } - // Transport attempt reached the network path; count as Ollama Cloud activity. + // Transport attempt reached the network path; count as Ollama Cloud / OpenCode Go activity. if s != nil { scheduleOllamaCloudUsageActivity(s.deferredService, account) + scheduleOpenCodeGoUsageActivity(s.deferredService, account) } if classifyOpenAITransportError(err).Persistent { diff --git a/backend/internal/service/opencode_go_usage.go b/backend/internal/service/opencode_go_usage.go index 88564c36d4cc..eba93d4db39a 100644 --- a/backend/internal/service/opencode_go_usage.go +++ b/backend/internal/service/opencode_go_usage.go @@ -2,11 +2,14 @@ package service import ( "context" + "crypto/sha256" "database/sql" + "encoding/hex" "encoding/json" "errors" "fmt" "io" + "maps" "math/rand/v2" "net/http" "net/url" @@ -26,12 +29,22 @@ const ( OpenCodeGoUsageAutoRefreshExtraKey = "opencode_go_usage_auto_refresh" OpenCodeGoUsageSnapshotExtraKey = "opencode_go_usage_snapshot" + // OpenCodeGoUsageMinFetchInterval is the hard floor between two successful + // fetches of the same group, mirroring the floor nextOpenCodeGoUsageDelay + // applies to next_refresh_at. Activity may bring a refresh forward to this + // bound but never past it. Exported so the repository can apply the same + // floor inside the SQL due filter. + OpenCodeGoUsageMinFetchInterval = opencodeGoUsageMinIntervalMinutes * time.Minute + opencodeGoUsageAPIURL = "https://opencode.ai/zen/go/v1/usage" opencodeGoUsageDefaultIntervalMinutes = 15 opencodeGoUsageMinIntervalMinutes = 5 opencodeGoUsageMaxIntervalMinutes = 24 * 60 + opencodeGoUsageDefaultDebounceMinutes = 1 + opencodeGoUsageMinDebounceMinutes = 1 + opencodeGoUsageMaxDebounceMinutes = 60 opencodeGoUsageCycleInterval = time.Minute - opencodeGoUsageManualRefreshInterval = 10 * time.Second + opencodeGoUsageManualRefreshInterval = 30 * time.Second opencodeGoUsageRequestTimeout = 15 * time.Second opencodeGoUsageMaxBodyBytes = 512 * 1024 opencodeGoUsageMaxPerCycle = 20 @@ -52,7 +65,7 @@ var ( "OPENCODE_GO_USAGE_IDENTITY_CHANGED", "account identity or proxy changed during refresh; retry", ) ErrOpenCodeGoUsageRefreshRateLimited = infraerrors.TooManyRequests( - "OPENCODE_GO_USAGE_REFRESH_RATE_LIMITED", "OpenCode Go usage can be refreshed manually once every 10 seconds", + "OPENCODE_GO_USAGE_REFRESH_RATE_LIMITED", "OpenCode Go usage can be refreshed manually once every 30 seconds", ) ) @@ -62,10 +75,15 @@ const ( OpenCodeGoUsageStatusFailed = "failed" ) -// OpenCodeGoUsageSettings controls the opt-in periodic refresh runner. +// OpenCodeGoUsageSettings controls the opt-in request-driven refresh runner. +// +// IntervalMinutes is the max-wait bound: when model requests keep arriving and +// the trailing debounce keeps sliding, a refresh is forced after this long. +// DebounceMinutes is the quiet period after the latest request in a group. type OpenCodeGoUsageSettings struct { Enabled bool `json:"enabled"` - IntervalMinutes int `json:"interval_minutes"` + IntervalMinutes int `json:"interval_minutes"` // max wait while requests continue + DebounceMinutes int `json:"debounce_minutes"` // trailing quiet period after last request } // OpenCodeGoUsageWindow is a narrow, sanitized view of one official usage window. @@ -83,6 +101,12 @@ type OpenCodeGoUsageData struct { } // OpenCodeGoUsageSnapshot is the only usage observation persisted in account extra. +// +// NextRefreshAt remains a persisted compatibility field. For status=ok it is a +// max-wait horizon marker only; automatic success refreshes are driven by model +// request activity (group last_used_at + debounce/max-wait), not by this field +// alone. For failed/unauthorized snapshots it is the failure not-before time +// (Retry-After / exponential backoff) and is enforced as max(activityDue, NextRefreshAt). type OpenCodeGoUsageSnapshot struct { Status string `json:"status"` Data *OpenCodeGoUsageData `json:"data,omitempty"` @@ -103,9 +127,11 @@ type OpenCodeGoUsageState struct { } type openCodeGoUsageRepository interface { + ListOpenCodeGoUsageGroupAccounts(context.Context, []*Account) ([]Account, error) SetOpenCodeGoUsageAutoRefresh(context.Context, *Account, bool) error UpdateOpenCodeGoUsageSnapshot(context.Context, *Account, *OpenCodeGoUsageSnapshot) error - ListDueOpenCodeGoUsageAccounts(context.Context, time.Time, int) ([]Account, error) + DisableOpenCodeGoUsageAutoRefresh(context.Context, *Account) error + ListDueOpenCodeGoUsageAccounts(context.Context, time.Time, time.Duration, time.Duration, int) ([]Account, error) } // GetOpenCodeGoUsageSettings returns fail-safe defaults when the setting is absent. @@ -131,6 +157,9 @@ func (s *SettingService) GetOpenCodeGoUsageSettings(ctx context.Context) (*OpenC if settings.IntervalMinutes == 0 { settings.IntervalMinutes = defaults.IntervalMinutes } + if settings.DebounceMinutes == 0 { + settings.DebounceMinutes = defaults.DebounceMinutes + } normalizeOpenCodeGoUsageSettings(&settings) return &settings, nil } @@ -142,12 +171,31 @@ func (s *SettingService) SetOpenCodeGoUsageSettings(ctx context.Context, setting if settings == nil { return infraerrors.BadRequest("INVALID_OPENCODE_GO_USAGE_SETTINGS", "settings cannot be nil") } + if settings.DebounceMinutes == 0 { + // Legacy clients that omit debounce_minutes keep the fail-safe default. + settings.DebounceMinutes = opencodeGoUsageDefaultDebounceMinutes + } if settings.IntervalMinutes < opencodeGoUsageMinIntervalMinutes || settings.IntervalMinutes > opencodeGoUsageMaxIntervalMinutes { return infraerrors.BadRequest( "INVALID_OPENCODE_GO_USAGE_INTERVAL", fmt.Sprintf("interval_minutes must be between %d and %d", opencodeGoUsageMinIntervalMinutes, opencodeGoUsageMaxIntervalMinutes), ) } + if settings.DebounceMinutes < opencodeGoUsageMinDebounceMinutes || settings.DebounceMinutes > opencodeGoUsageMaxDebounceMinutes { + return infraerrors.BadRequest( + "INVALID_OPENCODE_GO_USAGE_DEBOUNCE", + fmt.Sprintf("debounce_minutes must be between %d and %d", opencodeGoUsageMinDebounceMinutes, opencodeGoUsageMaxDebounceMinutes), + ) + } + // The due time is min(lastUsed+debounce, fetchedAt+maxWait). Once the debounce + // reaches the max wait the debounce term can never win, so the knob would be + // silently inert instead of doing what the operator asked for. + if settings.DebounceMinutes >= settings.IntervalMinutes { + return infraerrors.BadRequest( + "INVALID_OPENCODE_GO_USAGE_DEBOUNCE", + fmt.Sprintf("debounce_minutes (%d) must be less than interval_minutes (%d)", settings.DebounceMinutes, settings.IntervalMinutes), + ) + } normalizeOpenCodeGoUsageSettings(settings) data, err := json.Marshal(settings) if err != nil { @@ -160,6 +208,7 @@ func defaultOpenCodeGoUsageSettings() *OpenCodeGoUsageSettings { return &OpenCodeGoUsageSettings{ Enabled: false, IntervalMinutes: opencodeGoUsageDefaultIntervalMinutes, + DebounceMinutes: opencodeGoUsageDefaultDebounceMinutes, } } @@ -170,20 +219,125 @@ func normalizeOpenCodeGoUsageSettings(settings *OpenCodeGoUsageSettings) { if settings.IntervalMinutes > opencodeGoUsageMaxIntervalMinutes { settings.IntervalMinutes = opencodeGoUsageMaxIntervalMinutes } + if settings.DebounceMinutes <= 0 { + settings.DebounceMinutes = opencodeGoUsageDefaultDebounceMinutes + } + if settings.DebounceMinutes < opencodeGoUsageMinDebounceMinutes { + settings.DebounceMinutes = opencodeGoUsageMinDebounceMinutes + } + if settings.DebounceMinutes > opencodeGoUsageMaxDebounceMinutes { + settings.DebounceMinutes = opencodeGoUsageMaxDebounceMinutes + } +} + +func openCodeGoUsageDurations(settings *OpenCodeGoUsageSettings) (debounce, maxWait time.Duration) { + normalized := defaultOpenCodeGoUsageSettings() + if settings != nil { + *normalized = *settings + } + normalizeOpenCodeGoUsageSettings(normalized) + return time.Duration(normalized.DebounceMinutes) * time.Minute, + time.Duration(normalized.IntervalMinutes) * time.Minute } // openCodeGoUsageIsAutoRefreshDue decides whether a configured auto-refresh -// account should fetch now. Missing or invalid snapshots fail open to a first -// fetch; otherwise the next_refresh_at horizon (success interval or failure -// backoff) decides. -func openCodeGoUsageIsAutoRefreshDue(snapshot *OpenCodeGoUsageSnapshot, now time.Time) bool { +// group should fetch now. groupLastUsedAt must be MAX(last_used_at) across the +// exact api_key group so shared multi-account groups do not miss activity. +// +// Success: a request must be newer than fetched_at; dueAt = min(lastUsed+debounce, fetchedAt+maxWait). +// Failure: a request must be newer than last_attempt_at; activity due uses the same min formula, +// then dueAt = max(activityDue, next_refresh_at) so Retry-After / exponential backoff win. +// Missing or invalid snapshots fail open to a first fetch. +func openCodeGoUsageIsAutoRefreshDue( + snapshot *OpenCodeGoUsageSnapshot, + groupLastUsedAt *time.Time, + now time.Time, + debounce, maxWait time.Duration, +) bool { + dueAt, ok := openCodeGoUsageAutoRefreshDueAt(snapshot, groupLastUsedAt, debounce, maxWait) + if !ok { + return false + } + return !now.Before(dueAt) +} + +func openCodeGoUsageAutoRefreshDueAt( + snapshot *OpenCodeGoUsageSnapshot, + groupLastUsedAt *time.Time, + debounce, maxWait time.Duration, +) (time.Time, bool) { + if debounce <= 0 { + debounce = time.Duration(opencodeGoUsageDefaultDebounceMinutes) * time.Minute + } + if maxWait <= 0 { + maxWait = time.Duration(opencodeGoUsageDefaultIntervalMinutes) * time.Minute + } if snapshot == nil { - return true + return time.Time{}, true } - if snapshot.NextRefreshAt.IsZero() { - return true + switch snapshot.Status { + case OpenCodeGoUsageStatusOK: + if snapshot.FetchedAt == nil || snapshot.FetchedAt.IsZero() { + return time.Time{}, true + } + fetchedAt := snapshot.FetchedAt.UTC() + if groupLastUsedAt == nil || !groupLastUsedAt.After(fetchedAt) { + return time.Time{}, false + } + lastUsed := groupLastUsedAt.UTC() + dueAt := minTime(lastUsed.Add(debounce), fetchedAt.Add(maxWait)) + // Keep the pre-existing hard floor between successful fetches. The success + // path no longer consults next_refresh_at, which is where + // nextOpenCodeGoUsageDelay used to apply opencodeGoUsageMinIntervalMinutes; + // without this, request traffic spaced slightly wider than the debounce + // drives the group's outbound rate far above the previous minimum. + if floor := fetchedAt.Add(OpenCodeGoUsageMinFetchInterval); dueAt.Before(floor) { + return floor, true + } + return dueAt, true + case OpenCodeGoUsageStatusFailed, OpenCodeGoUsageStatusUnauthorized: + if snapshot.LastAttemptAt.IsZero() { + return time.Time{}, true + } + lastAttempt := snapshot.LastAttemptAt.UTC() + if groupLastUsedAt == nil || !groupLastUsedAt.After(lastAttempt) { + return time.Time{}, false + } + lastUsed := groupLastUsedAt.UTC() + activityDue := minTime(lastUsed.Add(debounce), lastAttempt.Add(maxWait)) + if !snapshot.NextRefreshAt.IsZero() && snapshot.NextRefreshAt.UTC().After(activityDue) { + return snapshot.NextRefreshAt.UTC(), true + } + return activityDue, true + default: + return time.Time{}, true + } +} + +// maxOpenCodeGoUsageGroupLastUsed returns the newest last_used_at among group members. +func maxOpenCodeGoUsageGroupLastUsed(accounts []Account) *time.Time { + var latest *time.Time + for i := range accounts { + candidate := accounts[i].LastUsedAt + if candidate == nil || candidate.IsZero() { + continue + } + if latest == nil || candidate.After(*latest) { + ts := candidate.UTC() + latest = &ts + } } - return !now.Before(snapshot.NextRefreshAt) + return latest +} + +// scheduleOpenCodeGoUsageActivity records that an OpenCode Go API-key account +// actually attempted an upstream model request (including 429/5xx/transport errors). +// Local auth/validation failures must not call this. DeferredService dedupes writes. +func scheduleOpenCodeGoUsageActivity(deferred *DeferredService, account *Account) { + if deferred == nil || account == nil || !IsOpenCodeGoUsageAccount(account) { + return + } + deferred.ScheduleLastUsedUpdate(account.ID) } // OpenCodeGoUsageService refreshes the official usage JSON without affecting routing state. @@ -308,9 +462,96 @@ func (s *OpenCodeGoUsageService) GetState(ctx context.Context, accountID int64) if err != nil { return nil, err } + if err := s.ResolveOpenCodeGoUsageAccounts(ctx, []*Account{account}); err != nil { + return nil, err + } return OpenCodeGoUsageStateFromAccount(account), nil } +// ResolveOpenCodeGoUsageAccounts overlays group-owned managed state onto the +// supplied account objects. The repository resolves all matching siblings in one +// bounded query, so account-list responses do not issue one query per row. +func (s *OpenCodeGoUsageService) ResolveOpenCodeGoUsageAccounts(ctx context.Context, accounts []*Account) error { + if s == nil || s.accountRepo == nil || len(accounts) == 0 { + return nil + } + writer, ok := s.accountRepo.(openCodeGoUsageRepository) + if !ok { + return nil + } + eligible := make([]*Account, 0, len(accounts)) + for _, account := range accounts { + if _, ok := openCodeGoUsageGroupFingerprint(account); ok { + eligible = append(eligible, account) + } + } + if len(eligible) == 0 { + return nil + } + siblings, err := writer.ListOpenCodeGoUsageGroupAccounts(ctx, eligible) + if err != nil { + return fmt.Errorf("resolve OpenCode Go usage groups: %w", err) + } + sources := make(map[string]*Account) + for index := range siblings { + candidate := &siblings[index] + fingerprint, valid := openCodeGoUsageGroupFingerprint(candidate) + if !valid { + continue + } + current := sources[fingerprint] + if current == nil || candidate.UpdatedAt.After(current.UpdatedAt) || + (candidate.UpdatedAt.Equal(current.UpdatedAt) && candidate.ID < current.ID) { + sources[fingerprint] = candidate + } + } + resolvedSources := make(map[string]*Account, len(sources)) + for fingerprint, source := range sources { + clone := *source + clone.Extra = make(map[string]any, len(source.Extra)) + maps.Copy(clone.Extra, source.Extra) + resolvedSources[fingerprint] = &clone + } + for index := range siblings { + candidate := &siblings[index] + fingerprint, valid := openCodeGoUsageGroupFingerprint(candidate) + source := resolvedSources[fingerprint] + if !valid || source == nil { + continue + } + candidateSnapshot := decodeOpenCodeGoUsageSnapshot(candidate.Extra) + currentSnapshot := decodeOpenCodeGoUsageSnapshot(source.Extra) + if candidateSnapshot != nil && (currentSnapshot == nil || candidateSnapshot.LastAttemptAt.After(currentSnapshot.LastAttemptAt)) { + source.Extra[OpenCodeGoUsageSnapshotExtraKey] = candidate.Extra[OpenCodeGoUsageSnapshotExtraKey] + } + } + for _, account := range eligible { + fingerprint, _ := openCodeGoUsageGroupFingerprint(account) + applyOpenCodeGoUsageManagedExtra(account, resolvedSources[fingerprint]) + } + return nil +} + +func applyOpenCodeGoUsageManagedExtra(target, source *Account) { + if target == nil { + return + } + if target.Extra == nil { + target.Extra = make(map[string]any) + } + for _, key := range []string{ + OpenCodeGoUsageAutoRefreshExtraKey, + OpenCodeGoUsageSnapshotExtraKey, + } { + delete(target.Extra, key) + if source != nil && source.Extra != nil { + if value, ok := source.Extra[key]; ok { + target.Extra[key] = value + } + } + } +} + func (s *OpenCodeGoUsageService) SetAutoRefresh(ctx context.Context, accountID int64, enabled bool) (*OpenCodeGoUsageState, error) { if s == nil || s.accountRepo == nil { return nil, ErrOpenCodeGoUsageUnavailable @@ -322,6 +563,9 @@ func (s *OpenCodeGoUsageService) SetAutoRefresh(ctx context.Context, accountID i if !IsOpenCodeGoUsageAccount(account) { return nil, ErrOpenCodeGoUsageAccountInvalid } + if err := s.ResolveOpenCodeGoUsageAccounts(ctx, []*Account{account}); err != nil { + return nil, err + } writer, ok := s.accountRepo.(openCodeGoUsageRepository) if !ok { return nil, ErrOpenCodeGoUsageUnavailable @@ -367,23 +611,38 @@ func (s *OpenCodeGoUsageService) RunDue(ctx context.Context) error { return ErrOpenCodeGoUsageUnavailable } now := s.currentTime() - accounts, err := writer.ListDueOpenCodeGoUsageAccounts(ctx, now, opencodeGoUsageMaxPerCycle) + debounce, maxWait := openCodeGoUsageDurations(settings) + accounts, err := writer.ListDueOpenCodeGoUsageAccounts(ctx, now, debounce, maxWait, opencodeGoUsageMaxPerCycle) if err != nil { return fmt.Errorf("list due OpenCode Go usage accounts: %w", err) } var group errgroup.Group + seenGroups := make(map[string]struct{}, len(accounts)) for index := range accounts { account := accounts[index] - if !account.IsActive() || !openCodeGoUsageAutoRefreshEnabled(&account) { + fingerprint, valid := openCodeGoUsageGroupFingerprint(&account) + if !valid || !account.IsActive() || !openCodeGoUsageAutoRefreshEnabled(&account) { + continue + } + if _, duplicate := seenGroups[fingerprint]; duplicate { continue } + seenGroups[fingerprint] = struct{}{} snapshot := decodeOpenCodeGoUsageSnapshot(account.Extra) - if !openCodeGoUsageIsAutoRefreshDue(snapshot, now) { + // ListDue stamps Account.LastUsedAt with the api_key group MAX(last_used_at). + if !openCodeGoUsageIsAutoRefreshDue(snapshot, account.LastUsedAt, now, debounce, maxWait) { continue } accountID := account.ID + expected := account group.Go(func() error { if _, refreshErr := s.refreshAccount(ctx, accountID, settings, true); refreshErr != nil { + if errors.Is(refreshErr, ErrOpenCodeGoUsageIdentityChanged) { + if disableErr := writer.DisableOpenCodeGoUsageAutoRefresh(ctx, &expected); disableErr != nil { + logger.LegacyPrintf("service.opencode_go_usage", "disable_auto_refresh_failed: account_id=%d err=%v", accountID, disableErr) + } + return nil + } logger.LegacyPrintf("service.opencode_go_usage", "refresh_due_failed: account_id=%d err=%v", accountID, refreshErr) } return nil @@ -400,14 +659,15 @@ func (s *OpenCodeGoUsageService) refreshAccount(ctx context.Context, accountID i settings = defaultOpenCodeGoUsageSettings() } intervalMinutes := settings.IntervalMinutes + debounce, maxWait := openCodeGoUsageDurations(settings) anchor, err := s.accountRepo.GetByID(ctx, accountID) if err != nil { return nil, err } - if !IsOpenCodeGoUsageAccount(anchor) { + key, valid := openCodeGoUsageGroupFingerprint(anchor) + if !valid { return nil, ErrOpenCodeGoUsageAccountInvalid } - key := strconv.FormatInt(accountID, 10) value, err, _ := s.refreshGroup.Do(key, func() (any, error) { select { case s.refreshSlots <- struct{}{}: @@ -419,9 +679,16 @@ func (s *OpenCodeGoUsageService) refreshAccount(ctx context.Context, accountID i if loadErr != nil { return nil, loadErr } - if !IsOpenCodeGoUsageAccount(account) { + currentKey, currentValid := openCodeGoUsageGroupFingerprint(account) + if !currentValid { return nil, ErrOpenCodeGoUsageAccountInvalid } + if currentKey != key { + return nil, ErrOpenCodeGoUsageIdentityChanged + } + if err := s.ResolveOpenCodeGoUsageAccounts(ctx, []*Account{account}); err != nil { + return nil, err + } if !requireEnabled { if snapshot := decodeOpenCodeGoUsageSnapshot(account.Extra); snapshot != nil && !snapshot.LastAttemptAt.IsZero() { retryAt := snapshot.LastAttemptAt.Add(opencodeGoUsageManualRefreshInterval) @@ -438,7 +705,21 @@ func (s *OpenCodeGoUsageService) refreshAccount(ctx context.Context, accountID i if !account.IsActive() || !openCodeGoUsageAutoRefreshEnabled(account) { return nil, nil } - if !openCodeGoUsageIsAutoRefreshDue(decodeOpenCodeGoUsageSnapshot(account.Extra), s.currentTime()) { + groupLastUsed := account.LastUsedAt + if writer, ok := s.accountRepo.(openCodeGoUsageRepository); ok { + siblings, listErr := writer.ListOpenCodeGoUsageGroupAccounts(ctx, []*Account{account}) + if listErr != nil { + // Fall back to this account's own last_used_at. That is a narrower + // activity signal than the group maximum, so the due check may skip a + // refresh it would otherwise have run; surface it rather than + // silently changing the due semantics. + logger.LegacyPrintf("service.opencode_go_usage", + "group_last_used_lookup_failed: account_id=%d err=%v", account.ID, listErr) + } else { + groupLastUsed = maxOpenCodeGoUsageGroupLastUsed(siblings) + } + } + if !openCodeGoUsageIsAutoRefreshDue(decodeOpenCodeGoUsageSnapshot(account.Extra), groupLastUsed, s.currentTime(), debounce, maxWait) { return nil, nil } } @@ -618,6 +899,27 @@ func isOpenCodeGoBaseURL(raw string) bool { return strings.EqualFold(strings.TrimSuffix(parsed.Path, "/"), "/zen/go/v1") } +func openCodeGoUsageIdentity(account *Account) map[string]any { + if !IsOpenCodeGoUsageAccount(account) { + return nil + } + apiKey, ok := account.Credentials["api_key"].(string) + if !ok || apiKey == "" { + return nil + } + return map[string]any{"host": "opencode.ai", "api_key": apiKey} +} + +func openCodeGoUsageGroupFingerprint(account *Account) (string, bool) { + identity := openCodeGoUsageIdentity(account) + if identity == nil { + return "", false + } + apiKey, _ := identity["api_key"].(string) + sum := sha256.Sum256([]byte("opencode.ai\x00" + apiKey)) + return hex.EncodeToString(sum[:]), true +} + func isExactOpenCodeGoUsageURL(parsed *url.URL) bool { return parsed != nil && parsed.Scheme == "https" && parsed.Host == "opencode.ai" && parsed.Path == "/zen/go/v1/usage" && parsed.User == nil && parsed.RawQuery == "" && parsed.Fragment == "" && parsed.RawPath == "" diff --git a/backend/internal/service/opencode_go_usage_test.go b/backend/internal/service/opencode_go_usage_test.go index 15065ea0b0bd..e173a4cc28cf 100644 --- a/backend/internal/service/opencode_go_usage_test.go +++ b/backend/internal/service/opencode_go_usage_test.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "reflect" "sync" "sync/atomic" "testing" @@ -22,12 +23,17 @@ const openCodeGoUsageFixture = `{"usage":{"rolling":{"status":"ok","percent":6," type openCodeGoUsageTestRepo struct { AccountRepository - mu sync.Mutex - accounts map[int64]*Account - due []Account + mu sync.Mutex + accounts map[int64]*Account + due []Account + groupResolveCalls atomic.Int64 + getByIDCalls atomic.Int64 + disableAutoCalls atomic.Int64 + disableAutoAttempt atomic.Int64 } func (r *openCodeGoUsageTestRepo) GetByID(_ context.Context, id int64) (*Account, error) { + r.getByIDCalls.Add(1) r.mu.Lock() defer r.mu.Unlock() account := r.accounts[id] @@ -52,35 +58,109 @@ func (r *openCodeGoUsageTestRepo) GetByIDs(_ context.Context, ids []int64) ([]*A return result, nil } +func (r *openCodeGoUsageTestRepo) ListOpenCodeGoUsageGroupAccounts(_ context.Context, anchors []*Account) ([]Account, error) { + r.groupResolveCalls.Add(1) + r.mu.Lock() + defer r.mu.Unlock() + wanted := make(map[string]struct{}, len(anchors)) + for _, anchor := range anchors { + if fingerprint, ok := openCodeGoUsageGroupFingerprint(anchor); ok { + wanted[fingerprint] = struct{}{} + } + } + result := make([]Account, 0, len(r.accounts)) + for _, account := range r.accounts { + fingerprint, ok := openCodeGoUsageGroupFingerprint(account) + if _, match := wanted[fingerprint]; !ok || !match { + continue + } + result = append(result, cloneOpenCodeGoUsageTestAccount(*account)) + } + return result, nil +} + func (r *openCodeGoUsageTestRepo) SetOpenCodeGoUsageAutoRefresh(_ context.Context, expected *Account, enabled bool) error { r.mu.Lock() defer r.mu.Unlock() - account := r.accounts[expected.ID] - if account == nil { - return ErrAccountNotFound + members, err := r.openCodeGoGroupMembersLocked(expected) + if err != nil { + return err } - if account.Extra == nil { - account.Extra = make(map[string]any) + for _, account := range members { + applyOpenCodeGoUsageTestManagedExtra(account, expected) + account.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = enabled } - account.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = enabled return nil } func (r *openCodeGoUsageTestRepo) UpdateOpenCodeGoUsageSnapshot(_ context.Context, expected *Account, snapshot *OpenCodeGoUsageSnapshot) error { r.mu.Lock() defer r.mu.Unlock() - account := r.accounts[expected.ID] - if account == nil { - return ErrAccountNotFound + members, err := r.openCodeGoGroupMembersLocked(expected) + if err != nil { + return err } - if account.Extra == nil { - account.Extra = make(map[string]any) + for _, account := range members { + applyOpenCodeGoUsageTestManagedExtra(account, expected) + account.Extra[OpenCodeGoUsageSnapshotExtraKey] = snapshot } - account.Extra[OpenCodeGoUsageSnapshotExtraKey] = snapshot return nil } -func (r *openCodeGoUsageTestRepo) ListDueOpenCodeGoUsageAccounts(_ context.Context, _ time.Time, limit int) ([]Account, error) { +func (r *openCodeGoUsageTestRepo) DisableOpenCodeGoUsageAutoRefresh(_ context.Context, expected *Account) error { + r.disableAutoAttempt.Add(1) + r.mu.Lock() + defer r.mu.Unlock() + members, err := r.openCodeGoGroupMembersLocked(expected) + if err != nil { + return err + } + for _, account := range members { + applyOpenCodeGoUsageTestManagedExtra(account, expected) + account.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = false + delete(account.Extra, OpenCodeGoUsageSnapshotExtraKey) + } + r.disableAutoCalls.Add(1) + return nil +} + +func (r *openCodeGoUsageTestRepo) openCodeGoGroupMembersLocked(expected *Account) ([]*Account, error) { + anchor := r.accounts[expected.ID] + if !sameOpenCodeGoUsageTestIdentity(anchor, expected) { + return nil, ErrOpenCodeGoUsageIdentityChanged + } + fingerprint, ok := openCodeGoUsageGroupFingerprint(expected) + if !ok { + return nil, ErrOpenCodeGoUsageAccountInvalid + } + members := make([]*Account, 0, len(r.accounts)) + for _, account := range r.accounts { + candidate, valid := openCodeGoUsageGroupFingerprint(account) + if valid && candidate == fingerprint { + if account.Extra == nil { + account.Extra = make(map[string]any) + } + members = append(members, account) + } + } + return members, nil +} + +func applyOpenCodeGoUsageTestManagedExtra(account, source *Account) { + for _, key := range []string{OpenCodeGoUsageAutoRefreshExtraKey, OpenCodeGoUsageSnapshotExtraKey} { + delete(account.Extra, key) + if value, ok := source.Extra[key]; ok { + account.Extra[key] = value + } + } +} + +func sameOpenCodeGoUsageTestIdentity(left, right *Account) bool { + return left != nil && right != nil && left.Platform == right.Platform && left.Type == right.Type && + reflect.DeepEqual(left.Credentials, right.Credentials) && reflect.DeepEqual(left.ProxyID, right.ProxyID) +} + +func (r *openCodeGoUsageTestRepo) ListDueOpenCodeGoUsageAccounts(_ context.Context, _ time.Time, _, _ time.Duration, limit int) ([]Account, error) { r.mu.Lock() defer r.mu.Unlock() if len(r.due) > 0 { @@ -107,13 +187,14 @@ func cloneOpenCodeGoUsageTestAccount(account Account) Account { } type openCodeGoUsageHTTPStub struct { - status int - body []byte - header http.Header - calls atomic.Int64 - lastRequest *http.Request - lastProxy string - mu sync.Mutex + status int + body []byte + header http.Header + calls atomic.Int64 + beforeResponse func(*http.Request) + lastRequest *http.Request + lastProxy string + mu sync.Mutex } func (s *openCodeGoUsageHTTPStub) Do(req *http.Request, proxyURL string, _ int64, _ int) (*http.Response, error) { @@ -122,6 +203,9 @@ func (s *openCodeGoUsageHTTPStub) Do(req *http.Request, proxyURL string, _ int64 s.lastRequest = req s.lastProxy = proxyURL s.mu.Unlock() + if s.beforeResponse != nil { + s.beforeResponse(req) + } status := s.status if status == 0 { status = http.StatusOK @@ -359,11 +443,264 @@ func TestOpenCodeGoUsageManualRefreshNotThrottledAfterWindow(t *testing.T) { } func TestOpenCodeGoUsageIsAutoRefreshDue(t *testing.T) { + debounce := time.Minute + maxWait := time.Hour + now := time.Date(2026, time.July, 25, 12, 0, 0, 0, time.UTC) + fetched := now.Add(-30 * time.Minute) + ptr := func(ts time.Time) *time.Time { return &ts } + + require.True(t, openCodeGoUsageIsAutoRefreshDue(nil, nil, now, debounce, maxWait), "missing snapshot first due") + require.True(t, openCodeGoUsageIsAutoRefreshDue(&OpenCodeGoUsageSnapshot{Status: "bogus"}, nil, now, debounce, maxWait), "invalid status first due") + + okSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, FetchedAt: ptr(fetched), + LastAttemptAt: fetched, NextRefreshAt: fetched.Add(maxWait), + } + require.False(t, openCodeGoUsageIsAutoRefreshDue(okSnap, nil, now, debounce, maxWait), "no request after success") + require.False(t, openCodeGoUsageIsAutoRefreshDue(okSnap, ptr(fetched), now, debounce, maxWait), "request not after fetched_at") + require.False(t, openCodeGoUsageIsAutoRefreshDue(okSnap, ptr(now.Add(-30*time.Second)), now, debounce, maxWait), "debounce not elapsed") + require.True(t, openCodeGoUsageIsAutoRefreshDue(okSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "single request quiet for debounce") + + // Continuous requests: last used is now, but max-wait from old fetch forces due. + oldFetched := now.Add(-2 * time.Hour) + oldSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, FetchedAt: ptr(oldFetched), + LastAttemptAt: oldFetched, NextRefreshAt: oldFetched.Add(maxWait), + } + require.True(t, openCodeGoUsageIsAutoRefreshDue(oldSnap, ptr(now), now, debounce, maxWait), "max-wait forces due while requests continue") + // First request after a very old snapshot is immediately due because fetched+maxWait is past. + require.True(t, openCodeGoUsageIsAutoRefreshDue(oldSnap, ptr(now.Add(-time.Second)), now, debounce, maxWait), "stale snapshot first request immediate") + + failSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusFailed, FetchedAt: ptr(fetched), + LastAttemptAt: now.Add(-10 * time.Minute), NextRefreshAt: now.Add(20 * time.Minute), + } + require.False(t, openCodeGoUsageIsAutoRefreshDue(failSnap, nil, now, debounce, maxWait), "failure without new request") + require.False(t, openCodeGoUsageIsAutoRefreshDue(failSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "failure blocked by backoff") + failSnap.NextRefreshAt = now.Add(-time.Second) + require.True(t, openCodeGoUsageIsAutoRefreshDue(failSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "failure after backoff with new request") + + require.True(t, openCodeGoUsageIsAutoRefreshDue(&OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, LastAttemptAt: now, + }, nil, now, debounce, maxWait), "ok without fetched_at fails open") +} + +// The success path stopped consulting next_refresh_at, which is where +// nextOpenCodeGoUsageDelay used to apply the minimum interval. Activity may pull +// a refresh forward only as far as that floor, otherwise request traffic spaced +// just wider than the debounce drives the group's outbound rate far above the +// pre-existing minimum. +func TestOpenCodeGoUsageAutoRefreshDueAtHonoursMinFetchInterval(t *testing.T) { + debounce := time.Minute + maxWait := time.Hour + now := time.Date(2026, time.July, 25, 12, 0, 0, 0, time.UTC) + ptr := func(ts time.Time) *time.Time { return &ts } + + // Debounce elapsed, but the last successful fetch is inside the floor. + recent := now.Add(-2 * time.Minute) + recentSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, FetchedAt: ptr(recent), LastAttemptAt: recent, + } + dueAt, ok := openCodeGoUsageAutoRefreshDueAt(recentSnap, ptr(now.Add(-time.Minute)), debounce, maxWait) + require.True(t, ok) + require.Equal(t, recent.Add(OpenCodeGoUsageMinFetchInterval), dueAt, + "due time must be clamped to fetched_at + min fetch interval") + require.False(t, openCodeGoUsageIsAutoRefreshDue(recentSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), + "debounce alone must not refresh within the min fetch interval") + + // Once the floor has passed the debounce governs again. + atFloor := now.Add(-OpenCodeGoUsageMinFetchInterval) + floorSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, FetchedAt: ptr(atFloor), LastAttemptAt: atFloor, + } + require.True(t, openCodeGoUsageIsAutoRefreshDue(floorSnap, ptr(now.Add(-2*time.Minute)), now, debounce, maxWait), + "past the floor a quiet debounce window is due") + + // The floor never delays a refresh that max-wait has already forced. + oldFetched := now.Add(-2 * time.Hour) + oldSnap := &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, FetchedAt: ptr(oldFetched), LastAttemptAt: oldFetched, + } + dueAt, ok = openCodeGoUsageAutoRefreshDueAt(oldSnap, ptr(now), debounce, maxWait) + require.True(t, ok) + require.Equal(t, oldFetched.Add(maxWait), dueAt, "max-wait due time must not be pushed out by the floor") +} + +func TestOpenCodeGoUsageGroupSharesStateAcrossSiblings(t *testing.T) { + source := openCodeGoUsageAccount(71) + source.Credentials["api_key"] = "shared-key" + source.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = true + source.Extra[OpenCodeGoUsageSnapshotExtraKey] = &OpenCodeGoUsageSnapshot{ + Status: OpenCodeGoUsageStatusOK, + Data: &OpenCodeGoUsageData{Rolling: OpenCodeGoUsageWindow{Status: "ok", Percent: 6}}, + } + source.UpdatedAt = time.Now().Add(-time.Minute) + sibling := openCodeGoUsageAccount(72) + sibling.Credentials = map[string]any{"base_url": "HTTPS://OPENCODE.AI/ZEN/GO/V1/", "api_key": "shared-key"} + sibling.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = true + sibling.UpdatedAt = time.Now() + different := openCodeGoUsageAccount(73) + different.Credentials["api_key"] = "different-key" + repo := &openCodeGoUsageTestRepo{accounts: map[int64]*Account{ + source.ID: source, sibling.ID: sibling, different.ID: different, + }} + svc := newOpenCodeGoUsageTestService(t, repo, &openCodeGoUsageHTTPStub{}, &upstreamBillingProbeSettingRepo{}) + + state, err := svc.GetState(context.Background(), sibling.ID) + require.NoError(t, err) + require.True(t, state.Eligible) + require.True(t, state.AutoRefreshEnabled) + require.Equal(t, 6.0, state.Snapshot.Data.Rolling.Percent) + + differentState, err := svc.GetState(context.Background(), different.ID) + require.NoError(t, err) + require.False(t, differentState.AutoRefreshEnabled) + require.Nil(t, differentState.Snapshot) + + newSibling := openCodeGoUsageAccount(74) + newSibling.Credentials = map[string]any{"base_url": "https://opencode.ai/zen/go/v1", "api_key": "shared-key"} + repo.mu.Lock() + repo.accounts[newSibling.ID] = newSibling + repo.mu.Unlock() + newState, err := svc.GetState(context.Background(), newSibling.ID) + require.NoError(t, err) + require.True(t, newState.AutoRefreshEnabled) + require.Equal(t, state.Snapshot, newState.Snapshot) + + before := repo.groupResolveCalls.Load() + require.NoError(t, svc.ResolveOpenCodeGoUsageAccounts(context.Background(), []*Account{source, sibling, different, newSibling})) + require.Equal(t, before+1, repo.groupResolveCalls.Load(), "one list batch must issue one group lookup") +} + +func TestOpenCodeGoUsageSetAutoRefreshAndSnapshotAreGroupScoped(t *testing.T) { + first := openCodeGoUsageAccount(81) + first.Credentials["api_key"] = "shared-key" + second := openCodeGoUsageAccount(82) + second.Credentials = map[string]any{"base_url": "https://opencode.ai/zen/go/v1/", "api_key": "shared-key"} + different := openCodeGoUsageAccount(83) + different.Credentials["api_key"] = "different-key" + repo := &openCodeGoUsageTestRepo{accounts: map[int64]*Account{ + first.ID: first, second.ID: second, different.ID: different, + }} + svc := newOpenCodeGoUsageTestService(t, repo, &openCodeGoUsageHTTPStub{}, &upstreamBillingProbeSettingRepo{}) + + state, err := svc.SetAutoRefresh(context.Background(), second.ID, true) + require.NoError(t, err) + require.True(t, state.AutoRefreshEnabled) + require.Equal(t, true, first.Extra[OpenCodeGoUsageAutoRefreshExtraKey]) + require.Equal(t, true, second.Extra[OpenCodeGoUsageAutoRefreshExtraKey]) + require.NotContains(t, different.Extra, OpenCodeGoUsageAutoRefreshExtraKey) + + // A snapshot write must not wipe the auto-refresh switch (pure merge). now := time.Now().UTC() - require.True(t, openCodeGoUsageIsAutoRefreshDue(nil, now)) - require.True(t, openCodeGoUsageIsAutoRefreshDue(&OpenCodeGoUsageSnapshot{}, now)) - require.True(t, openCodeGoUsageIsAutoRefreshDue(&OpenCodeGoUsageSnapshot{NextRefreshAt: now.Add(-time.Minute)}, now)) - require.False(t, openCodeGoUsageIsAutoRefreshDue(&OpenCodeGoUsageSnapshot{NextRefreshAt: now.Add(time.Minute)}, now)) + _, err = svc.Refresh(context.Background(), first.ID) + require.NoError(t, err) + require.Equal(t, true, first.Extra[OpenCodeGoUsageAutoRefreshExtraKey], "snapshot write must preserve auto_refresh") + require.Equal(t, true, second.Extra[OpenCodeGoUsageAutoRefreshExtraKey], "snapshot write must preserve auto_refresh on siblings") + require.NotNil(t, decodeOpenCodeGoUsageSnapshot(second.Extra), "snapshot must be shared with siblings") + require.Equal(t, decodeOpenCodeGoUsageSnapshot(first.Extra), decodeOpenCodeGoUsageSnapshot(second.Extra)) + require.NotNil(t, now) +} + +func TestOpenCodeGoUsageRefreshSingleflightAndRunnerDeduplicateSharedGroup(t *testing.T) { + first := openCodeGoUsageAccount(91) + first.Credentials["api_key"] = "shared-key" + first.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = true + second := openCodeGoUsageAccount(92) + second.Credentials = map[string]any{"base_url": "https://opencode.ai/zen/go/v1/", "api_key": "shared-key"} + second.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = true + repo := &openCodeGoUsageTestRepo{ + accounts: map[int64]*Account{first.ID: first, second.ID: second}, + due: []Account{*first, *second}, + } + settingsRepo := &upstreamBillingProbeSettingRepo{values: map[string]string{ + SettingKeyOpenCodeGoUsageSettings: `{"enabled":true,"interval_minutes":15}`, + }} + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + upstream := &openCodeGoUsageHTTPStub{body: []byte(openCodeGoUsageFixture), beforeResponse: func(*http.Request) { + once.Do(func() { close(started) }) + <-release + }} + svc := newOpenCodeGoUsageTestService(t, repo, upstream, settingsRepo) + + errs := make(chan error, 2) + go func() { _, err := svc.Refresh(context.Background(), first.ID); errs <- err }() + <-started + // The first caller is now parked in the stub, having loaded the account twice + // (once to build the group key, once inside the singleflight function). + loadsBeforeSecond := repo.getByIDCalls.Load() + go func() { _, err := svc.Refresh(context.Background(), second.ID); errs <- err }() + // Only release the first caller once the second one has loaded its own + // account, which happens immediately before it joins the singleflight group. + require.Eventually(t, func() bool { + return repo.getByIDCalls.Load() > loadsBeforeSecond + }, 5*time.Second, time.Millisecond, "the second caller must reach the singleflight group before the first is released") + close(release) + require.NoError(t, <-errs) + require.NoError(t, <-errs) + require.Equal(t, int64(1), upstream.calls.Load()) + require.NotNil(t, decodeOpenCodeGoUsageSnapshot(first.Extra)) + require.Equal(t, decodeOpenCodeGoUsageSnapshot(first.Extra), decodeOpenCodeGoUsageSnapshot(second.Extra)) + + delete(first.Extra, OpenCodeGoUsageSnapshotExtraKey) + delete(second.Extra, OpenCodeGoUsageSnapshotExtraKey) + upstream.beforeResponse = nil + require.NoError(t, svc.RunDue(context.Background())) + require.Equal(t, int64(2), upstream.calls.Load(), "RunDue must issue one request for the shared group") +} + +func TestOpenCodeGoUsageRefreshRejectsGroupChangeBeforeUpstreamRequest(t *testing.T) { + account := openCodeGoUsageAccount(94) + base := &openCodeGoUsageTestRepo{accounts: map[int64]*Account{account.ID: account}} + repo := &openCodeGoRefreshPreflightIdentityChangeRepo{openCodeGoUsageTestRepo: base} + upstream := &openCodeGoUsageHTTPStub{body: []byte(openCodeGoUsageFixture)} + svc := NewOpenCodeGoUsageService(repo, upstream, NewSettingService(&upstreamBillingProbeSettingRepo{}, nil)) + t.Cleanup(svc.Stop) + + _, err := svc.Refresh(context.Background(), account.ID) + + require.ErrorIs(t, err, ErrOpenCodeGoUsageIdentityChanged) + require.Zero(t, upstream.calls.Load()) + require.NotContains(t, account.Extra, OpenCodeGoUsageSnapshotExtraKey) +} + +type openCodeGoRefreshPreflightIdentityChangeRepo struct { + *openCodeGoUsageTestRepo + getCalls atomic.Int64 +} + +func (r *openCodeGoRefreshPreflightIdentityChangeRepo) GetByID(ctx context.Context, id int64) (*Account, error) { + if r.getCalls.Add(1) == 2 { + r.mu.Lock() + r.accounts[id].Credentials["api_key"] = "rotated-before-refresh" + r.mu.Unlock() + } + return r.openCodeGoUsageTestRepo.GetByID(ctx, id) +} + +func TestOpenCodeGoUsageRunnerDisablesAutoRefreshAfterIdentityError(t *testing.T) { + account := openCodeGoUsageAccount(14) + account.Extra[OpenCodeGoUsageAutoRefreshExtraKey] = true + missingProxyID := int64(99) + account.ProxyID = &missingProxyID + account.Proxy = nil + repo := &openCodeGoUsageTestRepo{accounts: map[int64]*Account{14: account}} + settingsRepo := &upstreamBillingProbeSettingRepo{values: map[string]string{ + SettingKeyOpenCodeGoUsageSettings: `{"enabled":true,"interval_minutes":15}`, + }} + upstream := &openCodeGoUsageHTTPStub{body: []byte(openCodeGoUsageFixture)} + svc := newOpenCodeGoUsageTestService(t, repo, upstream, settingsRepo) + + require.NoError(t, svc.RunDue(context.Background())) + require.Equal(t, int64(1), repo.disableAutoCalls.Load()) + require.Equal(t, false, account.Extra[OpenCodeGoUsageAutoRefreshExtraKey]) + require.Zero(t, upstream.calls.Load()) + + require.NoError(t, svc.RunDue(context.Background())) + require.Equal(t, int64(1), repo.disableAutoCalls.Load()) + require.Zero(t, upstream.calls.Load()) } func TestOpenCodeGoUsageRunDueRefreshesDueAccounts(t *testing.T) { @@ -438,6 +775,43 @@ func TestIsOpenCodeGoUsageAccount(t *testing.T) { require.False(t, IsOpenCodeGoUsageAccount(account)) } +func TestOpenCodeGoUsageGroupFingerprint(t *testing.T) { + first := openCodeGoUsageAccount(1) + first.Credentials["api_key"] = "shared-key" + second := openCodeGoUsageAccount(2) + second.Credentials = map[string]any{"base_url": "HTTPS://OPENCODE.AI/ZEN/GO/V1/", "api_key": "shared-key"} + third := openCodeGoUsageAccount(3) + third.Credentials["api_key"] = "other-key" + + firstFP, firstOK := openCodeGoUsageGroupFingerprint(first) + secondFP, secondOK := openCodeGoUsageGroupFingerprint(second) + thirdFP, thirdOK := openCodeGoUsageGroupFingerprint(third) + require.True(t, firstOK) + require.True(t, secondOK) + require.True(t, thirdOK) + require.Equal(t, firstFP, secondFP, "same api_key across base_url variants must share a group") + require.NotEqual(t, firstFP, thirdFP, "different api_key must not share a group") + + // ineligible accounts have no fingerprint + account := openCodeGoUsageAccount(4) + account.Credentials["base_url"] = "https://opencode.ai/v1" + _, ok := openCodeGoUsageGroupFingerprint(account) + require.False(t, ok) +} + +func TestScheduleOpenCodeGoUsageActivityOnlyForOpenCode(t *testing.T) { + deferred := &DeferredService{} + openCode := openCodeGoUsageAccount(1) + openCode.Credentials["api_key"] = "k" + other := openCodeGoUsageAccount(2) + other.Credentials["base_url"] = "https://opencode.ai/v1" + + scheduleOpenCodeGoUsageActivity(deferred, openCode) + scheduleOpenCodeGoUsageActivity(deferred, other) + scheduleOpenCodeGoUsageActivity(nil, openCode) + scheduleOpenCodeGoUsageActivity(deferred, nil) +} + func TestOpenCodeGoUsageSettingsDefaultOffAndValidation(t *testing.T) { repo := &upstreamBillingProbeSettingRepo{} settingsService := NewSettingService(repo, nil) @@ -445,6 +819,7 @@ func TestOpenCodeGoUsageSettingsDefaultOffAndValidation(t *testing.T) { require.NoError(t, err) require.False(t, settings.Enabled) require.Equal(t, 15, settings.IntervalMinutes) + require.Equal(t, 1, settings.DebounceMinutes) // below the minimum err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 1}) @@ -452,13 +827,40 @@ func TestOpenCodeGoUsageSettingsDefaultOffAndValidation(t *testing.T) { // above the maximum err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 2000}) require.Error(t, err) + // debounce out of range + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 30, DebounceMinutes: 61}) + require.Error(t, err) + // DebounceMinutes=0 (legacy omit) defaults to 1 on write. + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 30, DebounceMinutes: 0}) + require.NoError(t, err) + settings, err = settingsService.GetOpenCodeGoUsageSettings(context.Background()) + require.NoError(t, err) + require.Equal(t, 1, settings.DebounceMinutes) // valid update round-trips - err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 30}) + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 30, DebounceMinutes: 2}) require.NoError(t, err) settings, err = settingsService.GetOpenCodeGoUsageSettings(context.Background()) require.NoError(t, err) require.True(t, settings.Enabled) require.Equal(t, 30, settings.IntervalMinutes) + require.Equal(t, 2, settings.DebounceMinutes) + + // debounce >= interval would make the debounce term unreachable in + // min(lastUsed+debounce, fetchedAt+maxWait), silently ignoring the operator's + // setting, so it is rejected rather than accepted and dropped. + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 5, DebounceMinutes: 5}) + require.Error(t, err, "debounce equal to interval must be rejected") + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 5, DebounceMinutes: 60}) + require.Error(t, err, "debounce greater than interval must be rejected") + err = settingsService.SetOpenCodeGoUsageSettings(context.Background(), &OpenCodeGoUsageSettings{Enabled: true, IntervalMinutes: 6, DebounceMinutes: 5}) + require.NoError(t, err, "debounce below interval stays valid") + + // Legacy JSON without debounce_minutes defaults to 1. + repo.values[SettingKeyOpenCodeGoUsageSettings] = `{"enabled":true,"interval_minutes":45}` + settings, err = settingsService.GetOpenCodeGoUsageSettings(context.Background()) + require.NoError(t, err) + require.Equal(t, 45, settings.IntervalMinutes) + require.Equal(t, 1, settings.DebounceMinutes) } func TestOpenCodeGoUsageStateFromAccount(t *testing.T) { diff --git a/frontend/src/i18n/locales/en/admin/settings.ts b/frontend/src/i18n/locales/en/admin/settings.ts index 296e7bfe9d29..ad1fc32cdfc8 100644 --- a/frontend/src/i18n/locales/en/admin/settings.ts +++ b/frontend/src/i18n/locales/en/admin/settings.ts @@ -455,8 +455,10 @@ export default { description: 'Refresh usage windows reported by the upstream OpenCode Go account for individually opted-in accounts. Disabled by default.', enabled: 'Enable global automatic refresh', enabledHint: 'Only accounts with their own automatic refresh switch enabled are refreshed. Manual refresh remains available.', - intervalMinutes: 'Refresh interval (minutes)', - intervalHint: 'Range: 5–1440 minutes.', + intervalMinutes: 'Max wait while requests continue (minutes)', + intervalHint: 'Range: 5–1440 minutes. When continuous requests keep sliding the debounce, force a refresh after this wait.', + debounceMinutes: 'Quiet period after last request (minutes)', + debounceHint: 'Range: 1–60 minutes, and must be less than the refresh interval. Refresh after the latest model request has been quiet for this long.', saved: 'OpenCode Go usage refresh settings saved', saveFailed: 'Failed to save OpenCode Go usage refresh settings' }, diff --git a/frontend/src/i18n/locales/zh/admin/settings.ts b/frontend/src/i18n/locales/zh/admin/settings.ts index c71da82590e7..897a7f79ea77 100644 --- a/frontend/src/i18n/locales/zh/admin/settings.ts +++ b/frontend/src/i18n/locales/zh/admin/settings.ts @@ -448,8 +448,10 @@ export default { description: '刷新上游 OpenCode Go 账号上报的用量窗口；默认关闭，仅对单独开启的账号生效。', enabled: '启用全局自动刷新', enabledHint: '仅刷新账号自身也开启自动刷新的账号。手动刷新不受影响。', - intervalMinutes: '刷新间隔（分钟）', - intervalHint: '范围 5–1440 分钟。', + intervalMinutes: '请求持续时的最长等待（分钟）', + intervalHint: '范围 5–1440 分钟。请求持续不断导致 debounce 一直后移时，最晚在此时间强制刷新。', + debounceMinutes: '请求安静等待（分钟）', + debounceHint: '范围 1–60 分钟，且必须小于刷新间隔。最后一次模型请求安静满此时长后再抓取用量。', saved: 'OpenCode Go 用量刷新设置已保存', saveFailed: '保存 OpenCode Go 用量刷新设置失败' }, diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 4aaa348fc6ab..3177120a696c 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -1141,7 +1141,10 @@ export interface OpenCodeGoUsageState { export interface OpenCodeGoUsageSettings { enabled: boolean + /** Max wait while model requests keep arriving (minutes). */ interval_minutes: number + /** Trailing quiet period after the latest model request (minutes). */ + debounce_minutes: number } export interface Account { diff --git a/frontend/src/views/admin/SettingsView.vue b/frontend/src/views/admin/SettingsView.vue index 0cc619c1f675..82d7824037ac 100644 --- a/frontend/src/views/admin/SettingsView.vue +++ b/frontend/src/views/admin/SettingsView.vue @@ -4893,6 +4893,24 @@ />
+
+ + +

+ {{ t("admin.settings.opencodeGoUsage.debounceHint") }} +

+