fix(libraryctl): 支持 MinIO 源，并与 API 共用同一份 scope 映射

为什么：专用验收实例用 S3 兼容的 MinIO，而 CLI 的源发现只认 oss 环境变量， `STORAGE_TYPE=minio` 直接失败；同时 CLI 只读 routing bindings，API 读 LIBRARY_PRODUCTION_KB_*，两边指向不同知识库时，"封存的是哪个库"就说不清。怎么改：新增 objectStoreFromSourceEnv（oss/minio，端点/桶缺失即报错，不退回默认值）， scopeBindings 让环境变量优先、bindings 兜底。单测覆盖 minio http/https、oss 与未知类型。
2026-10-29 04:58:45 +08:00
parent 1f482e4b49
commit 23cc6c0afe
3 changed files with 166 additions and 14 deletions
@@ -26,6 +26,17 @@
 - `LibraryJobRecord.kind` 之前只有 `seal | sync`，恢复作业（restore）在类型上不存在；
  补上后页面不再把恢复作业当成未知类型。

+### Fixed
+
+- **`libraryctl` 只认 OSS 源**：`STORAGE_TYPE=minio` 的目标在发现阶段就被拒
+  （`unsupported source storage type "minio"`），S3 兼容的专用验收实例无法封存。
+  现在按 `MINIO_ENDPOINT / MINIO_USE_SSL / MINIO_ACCESS_KEY_ID / MINIO_SECRET_ACCESS_KEY /
+  MINIO_BUCKET_NAME` 映射成 `provider=minio` + path-style；读不到端点/桶就明确报错，
+  不退回默认值。
+- **CLI 与 API 的 scope→知识库 映射不一致**：API 用 `LIBRARY_PRODUCTION_KB_*`，CLI 只读
+  routing bindings，同一台机器上会封存"配置里那个库"而不是 API 的验收目标。现在环境变量
+  优先、读不到才退回 bindings。
+
 ### Notes

 - 本次真实验收（kgmaint 专用实例 + 原版 WeKnora API）与本快照之外的部署计划见
@@ -0,0 +1,67 @@
+package main
+
+import "testing"
+
+// 目标实例可以是 OSS，也可以是 S3 兼容的 MinIO（本机专用验收实例就是 minio）。
+// 只认 oss 时，任何 minio 目标都会在发现阶段直接失败，等于这条验收链只能在生产存储上跑。
+func TestObjectStoreFromSourceEnvAcceptsMinIO(t *testing.T) {
+	app := map[string]string{
+		"STORAGE_TYPE":            "minio",
+		"MINIO_ENDPOINT":          "langfuse-minio:9000",
+		"MINIO_USE_SSL":           "false",
+		"MINIO_ACCESS_KEY_ID":     "key",
+		"MINIO_SECRET_ACCESS_KEY": "secret",
+		"MINIO_BUCKET_NAME":       "kgmaint-1028a",
+		"MINIO_PATH_PREFIX":       "kgmaint/",
+	}
+	config, detail, err := objectStoreFromSourceEnv(app, 1<<20)
+	if err != nil {
+		t.Fatalf("minio: %v", err)
+	}
+	if config.Provider != "minio" || config.Endpoint != "http://langfuse-minio:9000" || config.Secure {
+		t.Fatalf("config: %+v", config)
+	}
+	if config.Bucket != "kgmaint-1028a" || config.AccessKeyID != "key" || config.SecretAccessKey != "secret" {
+		t.Fatalf("credentials mapping: %+v", config)
+	}
+	if detail == "" || config.MaxObjectBytes != 1<<20 {
+		t.Fatalf("detail=%q max=%d", detail, config.MaxObjectBytes)
+	}
+}
+
+func TestObjectStoreFromSourceEnvAcceptsMinIOOverTLS(t *testing.T) {
+	app := map[string]string{
+		"STORAGE_TYPE": "minio", "MINIO_ENDPOINT": "minio.internal:9000", "MINIO_USE_SSL": "true",
+		"MINIO_ACCESS_KEY": "k2", "MINIO_SECRET_KEY": "s2", "MINIO_BUCKET_NAME": "bucket",
+	}
+	config, _, err := objectStoreFromSourceEnv(app, 1)
+	if err != nil {
+		t.Fatalf("minio tls: %v", err)
+	}
+	if !config.Secure || config.Endpoint != "https://minio.internal:9000" || config.AccessKeyID != "k2" {
+		t.Fatalf("config: %+v", config)
+	}
+}
+
+func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) {
+	app := map[string]string{
+		"STORAGE_TYPE": "oss", "OSS_ENDPOINT": "oss-cn-shenzhen.aliyuncs.com", "OSS_REGION": "cn-shenzhen",
+		"OSS_ACCESS_KEY": "ak", "OSS_SECRET_KEY": "sk", "OSS_BUCKET_NAME": "prod",
+	}
+	config, _, err := objectStoreFromSourceEnv(app, 1)
+	if err != nil {
+		t.Fatalf("oss: %v", err)
+	}
+	if config.Provider != "oss" || config.Endpoint != "https://oss-cn-shenzhen.aliyuncs.com" || !config.Secure {
+		t.Fatalf("config: %+v", config)
+	}
+}
+
+func TestObjectStoreFromSourceEnvRejectsUnknownStorage(t *testing.T) {
+	if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "cos"}, 1); err == nil {
+		t.Fatal("expected error for unsupported storage type")
+	}
+	if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "minio"}, 1); err == nil {
+		t.Fatal("expected error when minio endpoint is missing")
+	}
+}
@@ -244,9 +244,6 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string
 	if app["DB_DRIVER"] != "postgres" {
 		return config, errors.New("source database driver is not postgres")
 	}
-	if app["STORAGE_TYPE"] != "oss" {
-		return config, fmt.Errorf("unsupported source storage type %q", app["STORAGE_TYPE"])
-	}
 	address, err := containerAddress(ctx, opts.sourcePG)
 	if err != nil {
 		return config, err
@@ -267,25 +264,102 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string
 	config.SearchIndex = app["ELASTICSEARCH_INDEX"]
 	config.Search = v071.SearchExporterConfig{BaseURL: "http://" + esAddress + ":9200", Index: app["ELASTICSEARCH_INDEX"],
 		AuthorizationToken: app["ELASTICSEARCH_PASSWORD"], PageSize: 500}
-	endpoint := app["OSS_ENDPOINT"]
-	if endpoint == "" {
-		return config, errors.New("source object endpoint is not configured")
+	config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.maxObjectBytes)
+	if err != nil {
+		return config, err
 	}
-	if !strings.Contains(endpoint, "://") {
-		endpoint = "https://" + endpoint
-	}
-	config.Object = v071.S3ObjectStoreConfig{Provider: "oss", Endpoint: endpoint, Region: app["OSS_REGION"],
-		AccessKeyID: app["OSS_ACCESS_KEY"], SecretAccessKey: app["OSS_SECRET_KEY"], Bucket: app["OSS_BUCKET_NAME"],
-		Secure: strings.HasPrefix(endpoint, "https://"), MaxObjectBytes: opts.maxObjectBytes}
-	config.ObjectDetail = "bucket=" + app["OSS_BUCKET_NAME"] + " region=" + app["OSS_REGION"]
 	return config, nil
 }

+// objectStoreFromSourceEnv 把源应用的存储环境映射成对象存储配置。
+//
+// 只认 OSS 时，S3 兼容的 MinIO 目标（本机专用验收实例）会在发现阶段就被拒，
+// 验收链只能跑在生产存储上——这不是安全性，而是能力缺口。两种都要能读，
+// 读不出端点/桶就明确报错，不退回任何默认值。
+func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) {
+	storage := strings.ToLower(strings.TrimSpace(app["STORAGE_TYPE"]))
+	var config v071.S3ObjectStoreConfig
+	var endpoint, bucket string
+	switch storage {
+	case "oss":
+		config = v071.S3ObjectStoreConfig{Provider: "oss", Region: app["OSS_REGION"],
+			AccessKeyID: app["OSS_ACCESS_KEY"], SecretAccessKey: app["OSS_SECRET_KEY"],
+			Bucket: app["OSS_BUCKET_NAME"], MaxObjectBytes: maxObjectBytes}
+		endpoint, bucket = app["OSS_ENDPOINT"], app["OSS_BUCKET_NAME"]
+		if !strings.Contains(endpoint, "://") {
+			endpoint = "https://" + endpoint
+		}
+	case "minio":
+		accessKey := firstNonEmpty(app["MINIO_ACCESS_KEY_ID"], app["MINIO_ACCESS_KEY"])
+		secretKey := firstNonEmpty(app["MINIO_SECRET_ACCESS_KEY"], app["MINIO_SECRET_KEY"])
+		config = v071.S3ObjectStoreConfig{Provider: "minio", Region: app["MINIO_REGION"],
+			AccessKeyID: accessKey, SecretAccessKey: secretKey, Bucket: app["MINIO_BUCKET_NAME"],
+			MaxObjectBytes: maxObjectBytes}
+		endpoint, bucket = app["MINIO_ENDPOINT"], app["MINIO_BUCKET_NAME"]
+		if strings.TrimSpace(endpoint) == "" {
+			return v071.S3ObjectStoreConfig{}, "", errors.New("source minio endpoint is not configured")
+		}
+		scheme := "http"
+		if parseBool(app["MINIO_USE_SSL"]) {
+			scheme = "https"
+		}
+		if !strings.Contains(endpoint, "://") {
+			endpoint = scheme + "://" + endpoint
+		}
+	default:
+		return v071.S3ObjectStoreConfig{}, "", fmt.Errorf("unsupported source storage type %q", app["STORAGE_TYPE"])
+	}
+	if strings.TrimSpace(endpoint) == "" {
+		return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint is not configured")
+	}
+	config.Endpoint = endpoint
+	config.Secure = strings.HasPrefix(endpoint, "https://")
+	config.ForcePathStyle = storage == "minio"
+	if config.Bucket == "" {
+		return v071.S3ObjectStoreConfig{}, "", errors.New("source object bucket is not configured")
+	}
+	return config, "provider=" + config.Provider + " bucket=" + bucket, nil
+}
+
+func firstNonEmpty(values ...string) string {
+	for _, value := range values {
+		if strings.TrimSpace(value) != "" {
+			return value
+		}
+	}
+	return ""
+}
+
+func parseBool(value string) bool {
+	switch strings.ToLower(strings.TrimSpace(value)) {
+	case "1", "true", "yes", "on":
+		return true
+	}
+	return false
+}
+
 func urlEscape(value string) string {
 	replacer := strings.NewReplacer("%", "%25", ":", "%3A", "@", "%40", "/", "%2F", "?", "%3F", "#", "%23", "&", "%26", " ", "%20")
 	return replacer.Replace(value)
 }

+// scopeBindings 返回 scope→知识库 映射。治理 API 用的是 LIBRARY_PRODUCTION_KB_*，
+// CLI 读的是治理库里的 routing bindings；同一台机器上两份不一致时，
+// CLI 抓到的会是"配置里那个库"而不是 API 认为的目标——所以环境变量优先，
+// 读不到才退回 bindings。两边讲同一份身份，验收才可比。
+func scopeBindings(ctx context.Context, container string) (map[string]string, error) {
+	binding, err := readRoutingBindings(ctx, container)
+	if err != nil {
+		return nil, err
+	}
+	for scope, key := range map[string]string{"internal": "LIBRARY_PRODUCTION_KB_INTERNAL", "external": "LIBRARY_PRODUCTION_KB_EXTERNAL"} {
+		if value := strings.TrimSpace(os.Getenv(key)); value != "" {
+			binding[scope] = value
+		}
+	}
+	return binding, nil
+}
+
 func readRoutingBindings(ctx context.Context, container string) (map[string]string, error) {
 	raw, err := dockerOutput(ctx, "exec", container, "psql", "-U", "governance", "-d", "governance", "-tAc",
 		"SELECT internal_kb_id||'|'||external_kb_id FROM routing_kb_bindings WHERE id='active'")
@@ -391,7 +465,7 @@ func openArchive(opts options) (*librarysnapshot.Archive, error) {

 func runStatus(ctx context.Context, opts options) error {
 	report := map[string]any{}
-	binding, bindingErr := readRoutingBindings(ctx, opts.governancePG)
+	binding, bindingErr := scopeBindings(ctx, opts.governancePG)
 	var config sourceConfig
 	var err error
 	if bindingErr != nil {