diff --git a/README.md b/README.md index 8378bc1b..0a7b0797 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,7 @@ sub2api-cn-relay-manager/ - [docs/OPENCLAW_EXTERNAL_VALIDATION.md](./docs/OPENCLAW_EXTERNAL_VALIDATION.md) —— OpenClaw 最后一跳真实使用验证 - [docs/PROJECT_STRUCTURE.md](./docs/PROJECT_STRUCTURE.md) —— 当前仓库目录职责说明 - [scripts/README.md](./scripts/README.md) —— 脚本目录分层说明与常用入口 +- [deploy/tksea-portal/admin-batch-import.html](./deploy/tksea-portal/admin-batch-import.html) —— 最小 batch-import 管理页 背景/设计文档： diff --git a/deploy/README.md b/deploy/README.md index 4b476565..83b2e970 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -8,6 +8,9 @@ - `tksea-portal/index.html` - `https://sub.tksea.top/portal/` 的静态页面源码 +- `tksea-portal/admin-batch-import.html` + - `https://sub.tksea.top/portal/admin-batch-import.html` 的最小管理页 + - 直接消费 `POST /api/batch-import/runs` 与 `GET /api/batch-import/runs/*` - `tksea-portal/nginx.sub.tksea.top.conf.example` - `sub.tksea.top` 上 portal 路由与代理示例 diff --git a/deploy/tksea-portal/admin-batch-import.html b/deploy/tksea-portal/admin-batch-import.html new file mode 100644 index 00000000..438d269f --- /dev/null +++ b/deploy/tksea-portal/admin-batch-import.html @@ -0,0 +1,902 @@ + + + + + + Batch Import 管理台 + + + +
+
+
+
Batch Import Admin
+

供应商批量导入管理页

+

+ 这个页面只做三件事：发起 batch import、查看 run 摘要、拉取 item 级复用结果。 + 后端仍然以现有 `POST /api/batch-import/runs` 与 `GET /api/batch-import/runs/*` 为准， + 页面不引入额外协议。 +

+
    +
  • 直接展示 `matched_account_state`
    • +
  • 直接展示 `account_resolution`
    • +
  • 复用 / 快速启用 / 替换 一眼可见
    • +
+
+ +
+ +
+
+

发起导入

+

+ 用 admin token 直接调用当前控制面的 batch-import API。 + `entries` 每行一个供应商帐号：`base_url|api_key|model_a,model_b`。 +

+ +
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ +
+ `self_service` 会直接用这把 key 执行 gateway completion 验证。 +
+
+ + + + + +
+ + + +
+ +
等待操作。
+
+ +
+

Run 结果

+

+ 创建完成后会自动查询 run 摘要和 item 列表。也可以手动输入 run id 重新拉取。 +

+ +
+ +
+ + +
+
+ +
+ +
+
总条目0
+
完成0
+
Active0
+
Degraded0
+
Broken0
+
+ +
+ + + + +
+ +
+ + + + + + + + + + + + + + + +
ProviderBase URLSmoke ModelMatched / ResolutionAccessBadgesAdvisory
还没有结果。
+
+
+
+
+ + + + diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index e3b5ec5a..7d441c56 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -58,6 +58,11 @@ SUB2API_CRM_ADMIN_TOKEN=change-me-before-production SUB2API_CRM_LISTEN_ADDR=127. 当前正式入口： - `https://sub.tksea.top/portal/` +- `https://sub.tksea.top/portal/admin-batch-import.html` + - 最小管理页 + - 直接消费 `POST /api/batch-import/runs` + - 直接消费 `GET /api/batch-import/runs/{run_id}` + - 直接消费 `GET /api/batch-import/runs/{run_id}/items` 兼容入口： diff --git a/docs/EXECUTION_BOARD.md b/docs/EXECUTION_BOARD.md index ced0a766..62f86f11 100644 --- a/docs/EXECUTION_BOARD.md +++ b/docs/EXECUTION_BOARD.md @@ -20,9 +20,16 @@ - 旧地址 `https://sub.tksea.top/kimi-portal/` 当前保留为 `302` 跳转，避免历史分享链接失效 - 站点资产与 Nginx 路由不再只存在 `/tmp` 临时文件，已收口进仓库： - `deploy/tksea-portal/index.html` + - `deploy/tksea-portal/admin-batch-import.html` - `deploy/tksea-portal/nginx.sub.tksea.top.conf.example` - `scripts/deploy/deploy_tksea_portal.sh` - 新页面已补齐登录态、用户信息、可绑定分组、活跃订阅、历史 key 列表，以及“新创建 key 对应分组/模型”的即时展示 + - 同轮已补最小 batch-import 管理页： + - 地址：`/portal/admin-batch-import.html` + - 直接消费 `POST /api/batch-import/runs` + - 直接消费 `GET /api/batch-import/runs/{run_id}` + - 直接消费 `GET /api/batch-import/runs/{run_id}/items` + - 用于验证 `matched_account_state / account_resolution / provision_reused` - 线上无副作用验收已确认： - `GET /portal/` 返回 `200` - `GET /kimi-portal/` 返回 `302 -> /portal/` @@ -65,7 +72,7 @@ - 当前主仓不再需要依赖历史临时 pack `openai-cn-pack-kimi-a7m` - `kimi-a7m` provider manifest 现在也开始承载 `host_overlays` 元数据；本地已把 `sub2api v0.1.129` 的 Kimi A7M runtime overlay 说明与 `.patch` 资产纳入 `packs/openai-cn-pack/overlays/` - 新增 `go run ./cmd/cli apply-host-overlay` 最小执行器；当前 pack 内命中的 overlay 已可直接生成 patched 宿主构建目录，不再只是 preview/import 阶段的提示信息 - - 2026-05-25 已继续把路线 A 推进到运行态层面： +- 2026-05-25 已继续把路线 A 推进到运行态层面： - 从 `/tmp/sub2api-clean` 的 clean worktree `HEAD` 导出 stock 源码，再用 `go run ./cmd/cli apply-host-overlay --provider-id kimi-a7m --host-version 0.1.129` 生成全新 patched 源码树 - 基于该 patched 源码树重建 `localhost/sub2api:patched-overlay-20260525-clean`，并在独立 Podman 网络里启动新的 Postgres / Redis / App fresh-host - `artifacts/real-host-acceptance/20260525_local_v0129_kimi_a7m_patched_overlay_image_freshhost_clean/21-summary.json` 已确认：`import_batch_status=succeeded`、`provider_status=active`、`latest_access_status=subscription_ready`、`completion_ok=true`、`completion_status=200` @@ -159,6 +166,18 @@ - `scripts/acceptance/import_remote43_provider.sh` 会直探 provider `base_url` 对应的 upstream `/models` 与 `/chat/completions` - 新增 `21-summary.json`，用于把 completion 失败自动分流成 `host_compatibility_gap` 或 `upstream_key_quota_issue` +- 2026-05-27 已把 V2 batch-import reuse runtime 真正接到 live action： + - `internal/app/batch_runtime.go` 现已接入 `InspectReuse` + - runtime reuse 查询优先命中既有 `import_run_items`，再回退到 legacy `import_batches / import_batch_items / managed_resources / providers` + - 兼容 V2 短指纹与 legacy 完整 sha256 指纹 + - live run 现在可真实产出 `matched_account_state / account_resolution / provision_reused` +- 2026-05-27 继续用 `/portal/admin-batch-import.html` 做真实页面操作验证，抓到了一个 live reuse 兼容缺口并已在本地修正： + - real remote43 样本 `https://api.53hk.cn/v1 + sk-4175...d776 + host=remote43-kimi-patched-auto2-18169` 首轮返回 `TOKEN_EXPIRED`，根因是 CRM 中持久化的宿主 bearer 已过期；刷新 host auth 后，item 已能恢复到 `access_status=active` + - 旧版 runtime 仍把同一条历史账号判成 `matched_account_state=none / account_resolution=created`，根因是 live runtime 的 normalized `provider_id`（如 `api-53hk-42797c06`）与 legacy pack provider id（如 `minimax-53hk`）不一致时，legacy reuse fallback 只按 `provider_id` 精确匹配 + - 当前已补 `base_url` fallback + `ProviderMatched` 策略信号：legacy lookup 会补查相同 `base_url` 的 provider，且“同 base_url + 同 key + family covered”现在可以真实收敛到 `reused/reactivated` + - 定向回归已通过：`go test ./internal/app -run 'TestBatchImportHTTP/(create run action reuses matched legacy account|create run action reuses legacy account when pack provider id differs from normalized runtime id)$' -count=1`、`go test ./internal/batch -run TestDecideReuse -count=1`、`go test ./internal/store/sqlite -run 'TestProvidersRepoListBy(BaseURL|BaseURLEmpty)$' -count=1` + - remote43 二次复验现已补证：更新后的 CRM 二进制已替换到 `18173` 控制面，真实 rerun `run_1779882868037300268` 已确认 item 从 `account_resolution=created` 收敛为 `account_resolution=reused`，并且 `provision_reused=true`、`access_status=active` + - 当前剩余的细节是：该 rerun item 的 `matched_account_state` 仍为 `none`，说明“reuse 命中后是否补出 active/disabled/deprecated state badge”仍可继续优化；但这不影响本轮要验证的 `created -> reused` 结果成立 11. patched CRM external validation 已完成 diff --git a/docs/PROJECT_STRUCTURE.md b/docs/PROJECT_STRUCTURE.md index d138fa83..a1e8ddb7 100644 --- a/docs/PROJECT_STRUCTURE.md +++ b/docs/PROJECT_STRUCTURE.md @@ -72,6 +72,8 @@ - `deploy/tksea-portal/index.html` - `sub.tksea.top/portal/` 静态页 +- `deploy/tksea-portal/admin-batch-import.html` + - `sub.tksea.top/portal/admin-batch-import.html` 最小管理页 - `deploy/tksea-portal/nginx.sub.tksea.top.conf.example` - 对应 Nginx 路由示例 diff --git a/internal/app/batch_runtime.go b/internal/app/batch_runtime.go index 15f18614..53c4e779 100644 --- a/internal/app/batch_runtime.go +++ b/internal/app/batch_runtime.go @@ -37,6 +37,11 @@ func (r batchImportRuntimeRunner) execute(ctx context.Context) (BatchImportRunCr ItemStore: r.store.ImportRunItems(), ProbeModels: probe.ProviderModels, ProbeCapabilities: probe.ProbeCapabilities, + InspectReuse: batchImportReuseInspector{ + store: r.store, + hostRow: r.hostRow, + currentRunID: runID, + }.Inspect, Provisioner: batchImportProvisioner{ store: r.store, hostRow: r.hostRow, diff --git a/internal/app/batch_runtime_reuse.go b/internal/app/batch_runtime_reuse.go new file mode 100644 index 00000000..6ac878e5 --- /dev/null +++ b/internal/app/batch_runtime_reuse.go @@ -0,0 +1,421 @@ +package app + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strconv" + "strings" + + "sub2api-cn-relay-manager/internal/batch" + "sub2api-cn-relay-manager/internal/pack" + "sub2api-cn-relay-manager/internal/probe" + "sub2api-cn-relay-manager/internal/store/sqlite" +) + +type batchImportReuseInspector struct { + store *sqlite.DB + hostRow sqlite.Host + currentRunID string +} + +func (i batchImportReuseInspector) Inspect(ctx context.Context, input batch.ReuseLookupInput) (batch.ReuseLookupResult, error) { + if i.store == nil { + return batch.ReuseLookupResult{}, fmt.Errorf("store is required") + } + if i.hostRow.ID <= 0 { + return batch.ReuseLookupResult{}, fmt.Errorf("host row is required") + } + + if reuse, ok, err := i.lookupPriorRunItem(ctx, input); err != nil { + return batch.ReuseLookupResult{}, err + } else if ok { + return reuse, nil + } + + return i.lookupLegacyImportBatch(ctx, input) +} + +func (i batchImportReuseInspector) lookupPriorRunItem(ctx context.Context, input batch.ReuseLookupInput) (batch.ReuseLookupResult, bool, error) { + runs, err := i.store.ImportRuns().List(ctx, 1000) + if err != nil { + return batch.ReuseLookupResult{}, false, err + } + + for _, run := range runs { + if strings.TrimSpace(run.HostID) != strings.TrimSpace(i.hostRow.HostID) { + continue + } + if strings.TrimSpace(run.RunID) == strings.TrimSpace(i.currentRunID) { + continue + } + items, err := i.store.ImportRunItems().ListByRunID(ctx, run.RunID) + if err != nil { + return batch.ReuseLookupResult{}, false, err + } + for _, item := range items { + if strings.TrimSpace(item.ProviderID) != strings.TrimSpace(input.ProviderID) { + continue + } + if !apiKeyFingerprintMatches(item.APIKeyFingerprint, input.APIKeyFingerprint) { + continue + } + return i.reuseFromRunItem(ctx, item) + } + } + + return batch.ReuseLookupResult{}, false, nil +} + +func (i batchImportReuseInspector) reuseFromRunItem(ctx context.Context, item sqlite.ImportRunItem) (batch.ReuseLookupResult, bool, error) { + modelMapping, err := i.loadExistingModelMapping(ctx, item.ProviderID) + if err != nil { + return batch.ReuseLookupResult{}, false, err + } + + reusedAccountID := int64(0) + if item.AccountID != nil { + reusedAccountID = *item.AccountID + } else if item.ReusedFromAccountID != nil { + reusedAccountID = *item.ReusedFromAccountID + } + + state := strings.TrimSpace(item.MatchedAccountState) + if state == "" { + state = string(batch.MatchedAccountStateNone) + } + + return batch.ReuseLookupResult{ + ProviderMatched: true, + ExistingProviderID: strings.TrimSpace(item.ProviderID), + ExistingAccessStatus: normalizeRunItemAccessStatus(item.AccessStatus), + ExistingCanonicalFamilys: parseStringArrayJSON(item.CanonicalFamiliesJSON), + MatchedAccountID: reusedAccountID, + MatchedAccountState: batch.MatchedAccountState(state), + ExistingModelMapping: modelMapping, + LegacyBatchID: item.LegacyBatchID, + }, true, nil +} + +func (i batchImportReuseInspector) lookupLegacyImportBatch(ctx context.Context, input batch.ReuseLookupInput) (batch.ReuseLookupResult, error) { + providers, err := i.lookupLegacyProviders(ctx, input) + if err != nil { + return batch.ReuseLookupResult{}, err + } + + type candidate struct { + provider sqlite.Provider + batch sqlite.ImportBatch + item sqlite.ImportBatchItem + resources []sqlite.ManagedResource + } + + var best *candidate + for _, providerRow := range providers { + batches, err := i.store.ImportBatches().ListByProviderIDAndHostID(ctx, providerRow.ID, i.hostRow.ID) + if err != nil { + return batch.ReuseLookupResult{}, err + } + for _, batchRow := range batches { + items, err := i.store.ImportBatchItems().GetByBatchID(ctx, batchRow.ID) + if err != nil { + return batch.ReuseLookupResult{}, err + } + for _, item := range items { + if !apiKeyFingerprintMatches(item.KeyFingerprint, input.APIKeyFingerprint) { + continue + } + resources, err := i.store.ManagedResources().GetByBatchID(ctx, batchRow.ID) + if err != nil { + return batch.ReuseLookupResult{}, err + } + best = &candidate{ + provider: providerRow, + batch: batchRow, + item: item, + resources: resources, + } + break + } + if best != nil && best.batch.ID == batchRow.ID { + break + } + } + if best != nil { + break + } + } + + if best == nil { + return batch.ReuseLookupResult{}, nil + } + + modelMapping, err := providerModelMapping(best.provider) + if err != nil { + return batch.ReuseLookupResult{}, err + } + canonicalFamilies, err := providerCanonicalFamilies(best.provider) + if err != nil { + return batch.ReuseLookupResult{}, err + } + + accountHostID, err := accountIDFromProbeSummary(best.item.ProbeSummaryJSON) + if err != nil { + return batch.ReuseLookupResult{}, err + } + + return batch.ReuseLookupResult{ + ProviderMatched: true, + ExistingProviderID: strings.TrimSpace(best.provider.ProviderID), + ExistingAccessStatus: normalizeLegacyBatchAccessStatus(best.batch.AccessStatus), + ExistingCanonicalFamilys: canonicalFamilies, + MatchedAccountID: resolveManagedAccountNumericID(accountHostID, best.resources), + MatchedAccountState: normalizeLegacyMatchedAccountState(best.item.AccountStatus, best.batch.AccessStatus), + ExistingModelMapping: modelMapping, + LegacyBatchID: int64Ptr(best.batch.ID), + }, nil +} + +func (i batchImportReuseInspector) lookupLegacyProviders(ctx context.Context, input batch.ReuseLookupInput) ([]sqlite.Provider, error) { + seen := make(map[int64]struct{}) + providers := make([]sqlite.Provider, 0) + appendUnique := func(rows []sqlite.Provider) { + for _, row := range rows { + if _, ok := seen[row.ID]; ok { + continue + } + seen[row.ID] = struct{}{} + providers = append(providers, row) + } + } + + if strings.TrimSpace(input.ProviderID) != "" { + rows, err := i.store.Providers().ListByProviderID(ctx, input.ProviderID) + if err != nil { + return nil, err + } + appendUnique(rows) + } + + if strings.TrimSpace(input.BaseURL) != "" { + rows, err := i.store.Providers().ListByBaseURL(ctx, strings.TrimSpace(input.BaseURL)) + if err != nil { + return nil, err + } + appendUnique(rows) + } + + return providers, nil +} + +func (i batchImportReuseInspector) loadExistingModelMapping(ctx context.Context, providerID string) (map[string]string, error) { + providers, err := i.store.Providers().ListByProviderID(ctx, providerID) + if err != nil { + return nil, err + } + if len(providers) == 0 { + return nil, nil + } + + for idx := len(providers) - 1; idx >= 0; idx-- { + providerRow := providers[idx] + batchRow, err := i.store.ImportBatches().GetLatestByProviderIDAndHostID(ctx, providerRow.ID, i.hostRow.ID) + if err != nil { + if err == sql.ErrNoRows { + continue + } + return nil, err + } + if batchRow.ID <= 0 { + continue + } + return providerModelMapping(providerRow) + } + + return nil, nil +} + +func providerModelMapping(providerRow sqlite.Provider) (map[string]string, error) { + type channelTemplatePayload struct { + ModelMapping map[string]string `json:"model_mapping"` + } + + var manifest pack.ProviderManifest + if strings.TrimSpace(providerRow.ManifestJSON) != "" && strings.TrimSpace(providerRow.ManifestJSON) != "{}" { + if err := json.Unmarshal([]byte(providerRow.ManifestJSON), &manifest); err != nil { + return nil, fmt.Errorf("decode provider manifest for %q: %w", providerRow.ProviderID, err) + } + if len(manifest.ChannelTemplate.ModelMapping) > 0 { + return cloneStringMap(manifest.ChannelTemplate.ModelMapping), nil + } + } + + if strings.TrimSpace(providerRow.ChannelTemplateJSON) != "" && strings.TrimSpace(providerRow.ChannelTemplateJSON) != "{}" { + var payload channelTemplatePayload + if err := json.Unmarshal([]byte(providerRow.ChannelTemplateJSON), &payload); err != nil { + return nil, fmt.Errorf("decode provider channel template for %q: %w", providerRow.ProviderID, err) + } + return cloneStringMap(payload.ModelMapping), nil + } + + return map[string]string{}, nil +} + +func providerCanonicalFamilies(providerRow sqlite.Provider) ([]string, error) { + models := make([]string, 0) + + var manifest pack.ProviderManifest + if strings.TrimSpace(providerRow.ManifestJSON) != "" && strings.TrimSpace(providerRow.ManifestJSON) != "{}" { + if err := json.Unmarshal([]byte(providerRow.ManifestJSON), &manifest); err != nil { + return nil, fmt.Errorf("decode provider manifest for %q: %w", providerRow.ProviderID, err) + } + models = append(models, manifest.DefaultModels...) + for _, mapped := range manifest.ChannelTemplate.ModelMapping { + models = append(models, mapped) + } + } + + models = append(models, parseStringArrayJSON(providerRow.DefaultModelsJSON)...) + + seen := make(map[string]struct{}, len(models)) + families := make([]string, 0, len(models)) + for _, modelID := range models { + canonical := probe.CanonicalModelFamily(modelID) + if canonical == "" { + continue + } + if _, ok := seen[canonical]; ok { + continue + } + seen[canonical] = struct{}{} + families = append(families, canonical) + } + return families, nil +} + +func normalizeRunItemAccessStatus(raw string) batch.AccessStatus { + switch strings.TrimSpace(raw) { + case string(batch.AccessStatusActive): + return batch.AccessStatusActive + case string(batch.AccessStatusDegraded): + return batch.AccessStatusDegraded + case string(batch.AccessStatusBroken): + return batch.AccessStatusBroken + default: + return batch.AccessStatusUnknown + } +} + +func normalizeLegacyBatchAccessStatus(raw string) batch.AccessStatus { + switch strings.TrimSpace(raw) { + case "subscription_ready", "self_service_ready", "fully_ready": + return batch.AccessStatusActive + case "degraded": + return batch.AccessStatusDegraded + case "broken": + return batch.AccessStatusBroken + default: + return batch.AccessStatusUnknown + } +} + +func normalizeLegacyMatchedAccountState(accountStatus string, batchAccessStatus string) batch.MatchedAccountState { + if normalizeLegacyBatchAccessStatus(batchAccessStatus) == batch.AccessStatusBroken { + return batch.MatchedAccountStateBroken + } + + switch strings.TrimSpace(accountStatus) { + case "passed", "warning": + return batch.MatchedAccountStateActive + case "disabled": + return batch.MatchedAccountStateDisabled + case "deprecated": + return batch.MatchedAccountStateDeprecated + case "failed": + return batch.MatchedAccountStateBroken + default: + return batch.MatchedAccountStateNone + } +} + +func accountIDFromProbeSummary(summaryJSON string) (string, error) { + if strings.TrimSpace(summaryJSON) == "" { + return "", nil + } + var payload map[string]any + if err := json.Unmarshal([]byte(summaryJSON), &payload); err != nil { + return "", err + } + accountID, _ := payload["account_id"].(string) + return strings.TrimSpace(accountID), nil +} + +func resolveManagedAccountNumericID(accountHostID string, resources []sqlite.ManagedResource) int64 { + accountHostID = strings.TrimSpace(accountHostID) + if accountHostID == "" { + return 0 + } + if numericID, err := strconv.ParseInt(accountHostID, 10, 64); err == nil && numericID > 0 { + return numericID + } + for _, resource := range resources { + if strings.TrimSpace(resource.ResourceType) != "account" { + continue + } + if strings.TrimSpace(resource.HostResourceID) == accountHostID { + return resource.ID + } + } + return 0 +} + +func apiKeyFingerprintMatches(stored string, lookup string) bool { + stored = normalizeFingerprint(stored) + lookup = normalizeFingerprint(lookup) + if stored == "" || lookup == "" { + return false + } + if stored == lookup { + return true + } + return strings.HasPrefix(stored, lookup) || strings.HasPrefix(lookup, stored) +} + +func normalizeFingerprint(raw string) string { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return "" + } + return strings.TrimPrefix(trimmed, "sha256:") +} + +func parseStringArrayJSON(raw string) []string { + values := []string{} + if strings.TrimSpace(raw) == "" { + return values + } + if err := json.Unmarshal([]byte(raw), &values); err != nil { + return []string{} + } + return values +} + +func cloneStringMap(input map[string]string) map[string]string { + if len(input) == 0 { + return map[string]string{} + } + cloned := make(map[string]string, len(input)) + for key, value := range input { + cloned[key] = value + } + return cloned +} + +func int64Ptr(value int64) *int64 { + if value <= 0 { + return nil + } + cloned := value + return &cloned +} diff --git a/internal/app/http_batch_import_test.go b/internal/app/http_batch_import_test.go index 8d8f7ca9..05405509 100644 --- a/internal/app/http_batch_import_test.go +++ b/internal/app/http_batch_import_test.go @@ -2,12 +2,16 @@ package app import ( "context" + "crypto/sha256" + "encoding/json" "fmt" "net/http" "net/http/httptest" "strings" "testing" + "sub2api-cn-relay-manager/internal/batch" + "sub2api-cn-relay-manager/internal/pack" "sub2api-cn-relay-manager/internal/store/sqlite" "sub2api-cn-relay-manager/internal/testutil" ) @@ -197,6 +201,120 @@ func TestBatchImportHTTP(t *testing.T) { t.Fatalf("item = %+v, want current_stage=done and access_status=active", items[0]) } }) + + t.Run("create run action reuses matched legacy account", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(newBatchImportActionStubServer(t)) + defer server.Close() + + dsn := testutil.SQLiteTestDSN(t, "reuse-state.db", true) + store := testutil.OpenSQLiteStore(t, dsn) + defer closeAppTestStore(t, store) + + hostPK := mustCreateBatchImportActionHost(t, store, server.URL) + packPK, providerPK := mustSeedLegacyBatchImportProvider(t, store, server.URL) + legacyBatchID := mustCreateLegacyReusableBatch(t, store, hostPK, packPK, providerPK, "entry-key", "account_1") + + action := buildCreateBatchImportRunAction(dsn) + result, err := action(context.Background(), CreateBatchImportRunRequest{ + HostID: "host-1", + Mode: "strict", + AccessMode: "self_service", + ConfirmWaitTimeoutSec: 1, + ProbeAPIKey: "gateway-key", + Entries: []BatchImportEntryRequest{{ + BaseURL: server.URL, + APIKey: "entry-key", + RequestedModels: []string{"kimi-k2.6"}, + }}, + }) + if err != nil { + t.Fatalf("buildCreateBatchImportRunAction() reuse error = %v", err) + } + if strings.TrimSpace(result.RunID) == "" { + t.Fatalf("result.RunID = %q, want non-empty", result.RunID) + } + + items, err := store.ImportRunItems().ListByRunID(context.Background(), result.RunID) + if err != nil { + t.Fatalf("ImportRunItems().ListByRunID() reuse error = %v", err) + } + if len(items) != 1 { + t.Fatalf("len(items) = %d, want 1", len(items)) + } + item := items[0] + if !item.ProvisionReused || item.AccountResolution != "reused" || item.MatchedAccountState != "active" { + t.Fatalf("reuse item = %+v, want provision_reused + reused + active", item) + } + if item.LegacyBatchID == nil || *item.LegacyBatchID != legacyBatchID { + t.Fatalf("LegacyBatchID = %v, want %d", item.LegacyBatchID, legacyBatchID) + } + if item.CurrentStage != "done" || item.AccessStatus != "active" { + t.Fatalf("reuse item final state = %+v, want done/active", item) + } + + batches, err := store.ImportBatches().ListByProviderIDAndHostID(context.Background(), providerPK, hostPK) + if err != nil { + t.Fatalf("ImportBatches().ListByProviderIDAndHostID() error = %v", err) + } + if len(batches) != 1 { + t.Fatalf("len(batches) = %d, want 1 legacy batch only", len(batches)) + } + }) + + t.Run("create run action reuses legacy account when pack provider id differs from normalized runtime id", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(newBatchImportActionStubServer(t)) + defer server.Close() + + dsn := testutil.SQLiteTestDSN(t, "reuse-baseurl-fallback.db", true) + store := testutil.OpenSQLiteStore(t, dsn) + defer closeAppTestStore(t, store) + + hostPK := mustCreateBatchImportActionHost(t, store, server.URL) + packPK, providerPK := mustSeedLegacyBatchImportProviderWithID(t, store, server.URL, "legacy-pack-provider") + legacyBatchID := mustCreateLegacyReusableBatch(t, store, hostPK, packPK, providerPK, "entry-key", "101") + + action := buildCreateBatchImportRunAction(dsn) + result, err := action(context.Background(), CreateBatchImportRunRequest{ + HostID: "host-1", + Mode: "strict", + AccessMode: "self_service", + ConfirmWaitTimeoutSec: 1, + ProbeAPIKey: "gateway-key", + Entries: []BatchImportEntryRequest{{ + BaseURL: server.URL, + APIKey: "entry-key", + RequestedModels: []string{"kimi-k2.6"}, + }}, + }) + if err != nil { + t.Fatalf("buildCreateBatchImportRunAction() base_url fallback reuse error = %v", err) + } + if strings.TrimSpace(result.RunID) == "" { + t.Fatalf("result.RunID = %q, want non-empty", result.RunID) + } + + items, err := store.ImportRunItems().ListByRunID(context.Background(), result.RunID) + if err != nil { + t.Fatalf("ImportRunItems().ListByRunID() base_url fallback error = %v", err) + } + if len(items) != 1 { + t.Fatalf("len(items) = %d, want 1", len(items)) + } + item := items[0] + if !item.ProvisionReused || item.AccountResolution != "reused" || item.MatchedAccountState != "active" { + t.Fatalf("reuse item = %+v, want provision_reused + reused + active", item) + } + if item.LegacyBatchID == nil || *item.LegacyBatchID != legacyBatchID { + t.Fatalf("LegacyBatchID = %v, want %d", item.LegacyBatchID, legacyBatchID) + } + if item.CurrentStage != "done" || item.AccessStatus != "active" { + t.Fatalf("reuse item final state = %+v, want done/active", item) + } + }) } func TestBatchImportWrapperFunctions(t *testing.T) { @@ -363,6 +481,21 @@ func newBatchImportActionStubServer(t *testing.T) http.Handler { } writeJSON(w, http.StatusOK, map[string]any{"data": map[string]any{"items": []map[string]any{{"id": "kimi-k2.6", "display_name": "Kimi K2.6", "type": "chat"}}}}) }) + mux.HandleFunc("/api/v1/admin/accounts/101/test", func(w http.ResponseWriter, r *http.Request) { + if !requireBatchImportActionAdminToken(t, w, r) { + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("event: result\n")) + _, _ = w.Write([]byte("data: {\"status\":\"passed\",\"message\":\"smoke passed\",\"ok\":true}\n\n")) + }) + mux.HandleFunc("/api/v1/admin/accounts/101/models", func(w http.ResponseWriter, r *http.Request) { + if !requireBatchImportActionAdminToken(t, w, r) { + return + } + writeJSON(w, http.StatusOK, map[string]any{"data": map[string]any{"items": []map[string]any{{"id": "kimi-k2.6", "display_name": "Kimi K2.6", "type": "chat"}}}}) + }) mux.HandleFunc("/api/v1/admin/subscriptions/assign", func(w http.ResponseWriter, r *http.Request) { if !requireBatchImportActionAdminToken(t, w, r) { return @@ -425,3 +558,142 @@ func requireBatchImportActionAdminToken(t *testing.T, w http.ResponseWriter, r * } return true } + +func mustCreateBatchImportActionHost(t *testing.T, store *sqlite.DB, baseURL string) int64 { + t.Helper() + + hostPK, err := store.Hosts().Create(context.Background(), sqlite.Host{ + HostID: "host-1", + BaseURL: baseURL, + HostVersion: "0.1.126", + CapabilityProbeJSON: "{}", + AuthType: "apikey", + AuthToken: "host-token", + }) + if err != nil { + t.Fatalf("Hosts().Create() error = %v", err) + } + return hostPK +} + +func mustSeedLegacyBatchImportProvider(t *testing.T, store *sqlite.DB, baseURL string) (int64, int64) { + t.Helper() + + return mustSeedLegacyBatchImportProviderWithID(t, store, baseURL, batch.NormalizeProviderID(baseURL)) +} + +func mustSeedLegacyBatchImportProviderWithID(t *testing.T, store *sqlite.DB, baseURL, providerID string) (int64, int64) { + t.Helper() + + packPK, err := store.Packs().Create(context.Background(), sqlite.Pack{ + PackID: "seed-pack", + Version: "1.0.0", + Checksum: "seed-pack@1.0.0", + Vendor: "test", + TargetHost: "sub2api", + ManifestJSON: `{"pack_id":"seed-pack","version":"1.0.0","target_host":"sub2api"}`, + }) + if err != nil { + t.Fatalf("Packs().Create() error = %v", err) + } + + providerManifest := pack.ProviderManifest{ + ProviderID: strings.TrimSpace(providerID), + DisplayName: "Legacy Reuse Provider", + BaseURL: baseURL, + Platform: "openai", + AccountType: "apikey", + DefaultModels: []string{"kimi-k2.6"}, + SmokeTestModel: "kimi-k2.6", + GroupTemplate: pack.GroupTemplate{ + Name: "legacy-group", + RateMultiplier: 1, + }, + ChannelTemplate: pack.ChannelTemplate{ + Name: "legacy-channel", + ModelMapping: map[string]string{"kimi-k2.6": "kimi-k2.6"}, + }, + PlanTemplate: pack.PlanTemplate{ + Name: "legacy-plan", + Price: 1, + ValidityDays: 30, + ValidityUnit: "day", + }, + Import: pack.ImportOptions{ + SupportsMultiKey: true, + SupportsStrict: true, + SupportsPartial: true, + }, + } + manifestJSON, err := json.Marshal(providerManifest) + if err != nil { + t.Fatalf("json.Marshal(providerManifest) error = %v", err) + } + defaultModelsJSON, err := json.Marshal(providerManifest.DefaultModels) + if err != nil { + t.Fatalf("json.Marshal(defaultModels) error = %v", err) + } + channelTemplateJSON, err := json.Marshal(providerManifest.ChannelTemplate) + if err != nil { + t.Fatalf("json.Marshal(channelTemplate) error = %v", err) + } + + providerPK, err := store.Providers().Create(context.Background(), sqlite.Provider{ + PackID: packPK, + ProviderID: providerManifest.ProviderID, + DisplayName: providerManifest.DisplayName, + BaseURL: providerManifest.BaseURL, + Platform: providerManifest.Platform, + AccountType: providerManifest.AccountType, + DefaultModelsJSON: string(defaultModelsJSON), + SmokeTestModel: providerManifest.SmokeTestModel, + ChannelTemplateJSON: string(channelTemplateJSON), + ManifestJSON: string(manifestJSON), + }) + if err != nil { + t.Fatalf("Providers().Create() error = %v", err) + } + return packPK, providerPK +} + +func mustCreateLegacyReusableBatch(t *testing.T, store *sqlite.DB, hostPK int64, packPK int64, providerPK int64, apiKey string, accountID string) int64 { + t.Helper() + + batchID, err := store.ImportBatches().Create(context.Background(), sqlite.ImportBatch{ + HostID: hostPK, + PackID: packPK, + ProviderID: providerPK, + Mode: "strict", + BatchStatus: "succeeded", + AccessStatus: "self_service_ready", + }) + if err != nil { + t.Fatalf("ImportBatches().Create() error = %v", err) + } + + if _, err := store.ImportBatchItems().Create(context.Background(), sqlite.ImportBatchItem{ + BatchID: batchID, + KeyFingerprint: fullSHA256Fingerprint(apiKey), + AccountStatus: "passed", + ProbeSummaryJSON: fmt.Sprintf(`{"account_id":"%s"}`, accountID), + }); err != nil { + t.Fatalf("ImportBatchItems().Create() error = %v", err) + } + + if _, err := store.ManagedResources().Create(context.Background(), sqlite.ManagedResource{ + BatchID: batchID, + HostID: hostPK, + ResourceType: "account", + HostResourceID: accountID, + ResourceName: "legacy-account", + }); err != nil { + t.Fatalf("ManagedResources().Create(account) error = %v", err) + } + + return batchID +} + +func fullSHA256Fingerprint(value string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(value))) + return fmt.Sprintf("sha256:%x", sum[:]) +} diff --git a/internal/batch/reuse_policy.go b/internal/batch/reuse_policy.go index a6fe4e0c..59321d72 100644 --- a/internal/batch/reuse_policy.go +++ b/internal/batch/reuse_policy.go @@ -7,6 +7,7 @@ import ( ) type ReuseInput struct { + ProviderMatched bool ProviderID string CanonicalModelFamilies []string MatchedAccountID int64 @@ -36,7 +37,7 @@ func DecideReuse(input ReuseInput) ReuseDecision { decision.MatchedAccountState = MatchedAccountStateNone } - if !sameProvider(input.ProviderID, input.ExistingProviderID) || !decision.FamilyCovered { + if !providerMatched(input) || !decision.FamilyCovered { return decision } @@ -93,3 +94,10 @@ func canonicalFamiliesCovered(requested []string, existing []string) bool { func sameProvider(left, right string) bool { return strings.TrimSpace(left) != "" && strings.TrimSpace(left) == strings.TrimSpace(right) } + +func providerMatched(input ReuseInput) bool { + if input.ProviderMatched { + return true + } + return sameProvider(input.ProviderID, input.ExistingProviderID) +} diff --git a/internal/batch/reuse_policy_test.go b/internal/batch/reuse_policy_test.go index c6d49c96..4e4a8011 100644 --- a/internal/batch/reuse_policy_test.go +++ b/internal/batch/reuse_policy_test.go @@ -9,6 +9,7 @@ func TestDecideReuse(t *testing.T) { t.Parallel() decision := DecideReuse(ReuseInput{ + ProviderMatched: true, ProviderID: "api-deepseek-12345678", CanonicalModelFamilies: []string{"kimi-k2.6"}, MatchedAccountID: 101, @@ -38,6 +39,7 @@ func TestDecideReuse(t *testing.T) { t.Parallel() decision := DecideReuse(ReuseInput{ + ProviderMatched: true, ProviderID: "api-kimi-12345678", CanonicalModelFamilies: []string{"kimi-k2.6"}, MatchedAccountID: 202, @@ -61,6 +63,7 @@ func TestDecideReuse(t *testing.T) { t.Parallel() brokenProvider := DecideReuse(ReuseInput{ + ProviderMatched: true, ProviderID: "api-deepseek-12345678", CanonicalModelFamilies: []string{"deepseek-v4-pro"}, MatchedAccountState: MatchedAccountStateActive, @@ -76,6 +79,7 @@ func TestDecideReuse(t *testing.T) { } brokenAccount := DecideReuse(ReuseInput{ + ProviderMatched: true, ProviderID: "api-deepseek-12345678", CanonicalModelFamilies: []string{"deepseek-v4-pro"}, MatchedAccountState: MatchedAccountStateBroken, @@ -95,6 +99,7 @@ func TestDecideReuse(t *testing.T) { t.Parallel() decision := DecideReuse(ReuseInput{ + ProviderMatched: true, ProviderID: "api-kimi-12345678", CanonicalModelFamilies: []string{"kimi-k2.6"}, MatchedAccountState: MatchedAccountStateActive, @@ -110,4 +115,26 @@ func TestDecideReuse(t *testing.T) { t.Fatal("FamilyCovered = false, want true") } }) + + t.Run("base url matched legacy provider is reused even when provider ids differ", func(t *testing.T) { + t.Parallel() + + decision := DecideReuse(ReuseInput{ + ProviderMatched: true, + ProviderID: "api-53hk-42797c06", + CanonicalModelFamilies: []string{"minimax-m2.7-highspeed"}, + MatchedAccountID: 101, + MatchedAccountState: MatchedAccountStateActive, + ExistingProviderID: "minimax-53hk", + ExistingAccessStatus: AccessStatusActive, + ExistingCanonicalFamilys: []string{"minimax-m2.5-highspeed", "minimax-m2.7-highspeed"}, + }) + + if !decision.ProvisionReused { + t.Fatal("ProvisionReused = false, want true") + } + if decision.AccountResolution != AccountResolutionReused { + t.Fatalf("AccountResolution = %q, want %q", decision.AccountResolution, AccountResolutionReused) + } + }) } diff --git a/internal/batch/service.go b/internal/batch/service.go index 1008aa11..457090f3 100644 --- a/internal/batch/service.go +++ b/internal/batch/service.go @@ -53,12 +53,14 @@ type ReuseLookupInput struct { } type ReuseLookupResult struct { + ProviderMatched bool ExistingProviderID string ExistingAccessStatus AccessStatus ExistingCanonicalFamilys []string MatchedAccountID int64 MatchedAccountState MatchedAccountState ExistingModelMapping map[string]string + LegacyBatchID *int64 } type ProvisionRequest struct { @@ -201,6 +203,7 @@ func (s BatchImportService) StartRun(ctx context.Context, req BatchImportRunRequ } reuseDecision := DecideReuse(ReuseInput{ + ProviderMatched: reuseLookup.ProviderMatched, ProviderID: providerID, CanonicalModelFamilies: canonicalFamilies, MatchedAccountID: reuseLookup.MatchedAccountID, @@ -231,6 +234,8 @@ func (s BatchImportService) StartRun(ctx context.Context, req BatchImportRunRequ ProvisionReused: reuseDecision.ProvisionReused, ReusedFromProviderID: reuseDecision.ReusedFromProviderID, ReusedFromAccountID: int64PtrIfSet(reuseDecision.ReusedFromAccountID), + LegacyBatchID: reuseLookup.LegacyBatchID, + LegacyProviderID: strings.TrimSpace(reuseLookup.ExistingProviderID), } if reuseDecision.ProvisionReused { diff --git a/internal/store/sqlite/providers_repo.go b/internal/store/sqlite/providers_repo.go index de9f3f70..b7ed7bbd 100644 --- a/internal/store/sqlite/providers_repo.go +++ b/internal/store/sqlite/providers_repo.go @@ -112,6 +112,47 @@ func (r *ProvidersRepo) ListByProviderID(ctx context.Context, providerID string) return providers, nil } +func (r *ProvidersRepo) ListByBaseURL(ctx context.Context, baseURL string) ([]Provider, error) { + baseURL = strings.TrimSpace(baseURL) + if baseURL == "" { + return nil, fmt.Errorf("base_url is required") + } + + rows, err := r.db.QueryContext(ctx, `SELECT id, pack_id, provider_id, display_name, base_url, platform, account_type, default_models_json, smoke_test_model, group_template_json, channel_template_json, plan_template_json, import_options_json, manifest_json FROM providers WHERE base_url = ? ORDER BY id`, baseURL) + if err != nil { + return nil, fmt.Errorf("query providers by base_url %q: %w", baseURL, err) + } + defer rows.Close() + + providers := make([]Provider, 0) + for rows.Next() { + var provider Provider + if err := rows.Scan( + &provider.ID, + &provider.PackID, + &provider.ProviderID, + &provider.DisplayName, + &provider.BaseURL, + &provider.Platform, + &provider.AccountType, + &provider.DefaultModelsJSON, + &provider.SmokeTestModel, + &provider.GroupTemplateJSON, + &provider.ChannelTemplateJSON, + &provider.PlanTemplateJSON, + &provider.ImportOptionsJSON, + &provider.ManifestJSON, + ); err != nil { + return nil, fmt.Errorf("scan provider by base_url %q: %w", baseURL, err) + } + providers = append(providers, provider) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate providers by base_url %q: %w", baseURL, err) + } + return providers, nil +} + func (r *ProvidersRepo) GetByPackIDAndProviderID(ctx context.Context, packID int64, providerID string) (Provider, error) { if packID <= 0 { return Provider{}, fmt.Errorf("pack_id is required") diff --git a/internal/store/sqlite/providers_repo_test.go b/internal/store/sqlite/providers_repo_test.go index 64cdf904..089a2390 100644 --- a/internal/store/sqlite/providers_repo_test.go +++ b/internal/store/sqlite/providers_repo_test.go @@ -55,6 +55,24 @@ func TestProvidersRepoListByProviderID(t *testing.T) { } } +func TestProvidersRepoListByBaseURL(t *testing.T) { + store := openTestDB(t) + + packID1 := createTestPackWithSuffix(t, store, "base-a") + packID2 := createTestPackWithSuffix(t, store, "base-b") + + store.Providers().Create(context.Background(), Provider{PackID: packID1, ProviderID: "minimax-53hk", DisplayName: "MM1", BaseURL: "https://api.53hk.cn/v1", Platform: "openai"}) + store.Providers().Create(context.Background(), Provider{PackID: packID2, ProviderID: "api-53hk-42797c06", DisplayName: "MM2", BaseURL: "https://api.53hk.cn/v1", Platform: "openai"}) + + providers, err := store.Providers().ListByBaseURL(context.Background(), "https://api.53hk.cn/v1") + if err != nil { + t.Fatalf("ListByBaseURL() error = %v", err) + } + if len(providers) != 2 { + t.Fatalf("ListByBaseURL() count = %d, want 2", len(providers)) + } +} + func TestProvidersRepoListByPackID(t *testing.T) { store := openTestDB(t) packID := createTestPack(t, store) @@ -105,6 +123,18 @@ func TestProvidersRepoListByProviderIDEmpty(t *testing.T) { } } +func TestProvidersRepoListByBaseURLEmpty(t *testing.T) { + store := openTestDB(t) + + providers, err := store.Providers().ListByBaseURL(context.Background(), "https://missing.example.com/v1") + if err != nil { + t.Fatalf("ListByBaseURL() error = %v", err) + } + if len(providers) != 0 { + t.Fatalf("ListByBaseURL() count = %d, want 0", len(providers)) + } +} + func TestProvidersRepoUpsertCreatesNew(t *testing.T) { store := openTestDB(t) packID := createTestPack(t, store) diff --git a/scripts/deploy/deploy_tksea_portal.sh b/scripts/deploy/deploy_tksea_portal.sh index 0c7b9257..bcbf2cdc 100755 --- a/scripts/deploy/deploy_tksea_portal.sh +++ b/scripts/deploy/deploy_tksea_portal.sh @@ -7,7 +7,7 @@ REMOTE="${REMOTE:-ubuntu@43.155.133.187}" REMOTE_PORTAL_DIR="${REMOTE_PORTAL_DIR:-/var/www/sub2api-portal}" REMOTE_NGINX_SITE="${REMOTE_NGINX_SITE:-/etc/nginx/sites-available/tksea}" REMOTE_HOST_PORT="${REMOTE_HOST_PORT:-18169}" -LOCAL_PORTAL_INDEX="${LOCAL_PORTAL_INDEX:-$ROOT_DIR/deploy/tksea-portal/index.html}" +LOCAL_PORTAL_DIR="${LOCAL_PORTAL_DIR:-$ROOT_DIR/deploy/tksea-portal}" REMOTE_STAGE_DIR="${REMOTE_STAGE_DIR:-/tmp/sub2api-portal-deploy}" DRY_RUN="${DRY_RUN:-0}" @@ -43,18 +43,20 @@ main() { require_cmd ssh require_cmd scp - [[ -f "$LOCAL_PORTAL_INDEX" ]] || die "missing portal index: $LOCAL_PORTAL_INDEX" + [[ -d "$LOCAL_PORTAL_DIR" ]] || die "missing portal dir: $LOCAL_PORTAL_DIR" + [[ -f "$LOCAL_PORTAL_DIR/index.html" ]] || die "missing portal index: $LOCAL_PORTAL_DIR/index.html" if [[ "$DRY_RUN" != "1" ]]; then [[ -f "$KEY" ]] || die "missing ssh key: $KEY" fi - local tmpdir patch_file index_copy + local tmpdir patch_file portal_stage_dir tmpdir="$(mktemp -d)" trap "rm -rf $(printf '%q' "$tmpdir")" EXIT patch_file="$tmpdir/patch_tksea_portal_nginx.py" - index_copy="$tmpdir/index.html" + portal_stage_dir="$tmpdir/portal" - cp "$LOCAL_PORTAL_INDEX" "$index_copy" + mkdir -p "$portal_stage_dir" + cp -R "$LOCAL_PORTAL_DIR/." "$portal_stage_dir/" cat > "$patch_file" <