feat(release): 发布目标知识库可选(源库 → 目标库映射)

- 迁移 000044:production_kb_bindings + release_item_executions.target_knowledge_base_id
- 系统设置新增「生产知识库绑定(对内/对外)」:从生产 WeKnora 实际列表选库,
  保存时真校验(不存在拒收;生产端不可达只提示并如实回报),仅管理员可改
- 执行计划按绑定解析目标库并冻结进 release_item_executions;空目标直接拒收;
  已发布过的条目保留原目标(重试/回滚回到同一处)
- DocumentBundle.RemapKnowledgeBase:knowledges/chunks/knowledge_tags/wiki_pages/
  wiki_folders 的 knowledge_base_id 与 ES 文档字段改写为目标库,重算表/文档摘要并重新封装
- 契约比对改为「声明的映射对」(KnowledgeBaseMapping),其余字段继续 fail-closed
- 契约检查器 / 目标身份 / 验证器全部走映射后的目标库;对象通道无需改写
- Compose 拆分:数据面接线 compose.release-sim.yaml + 两道门禁 compose.release-gates.yaml
- 测试:RemapKnowledgeBase 全通道/无操作/fail-closed、映射契约、绑定接口、RBAC、真实 PG 集成
This commit is contained in:
姚章浩
2026-10-08 15:41:13 +08:00
parent cddc4c90d1
commit 6440e3f679
40 changed files with 1823 additions and 92 deletions
+68
View File
@@ -1289,6 +1289,74 @@
}
}
},
"/api/v1/production-kb-bindings": {
"get": {
"tags": [
"Settings"
],
"operationId": "getProductionBindings",
"summary": "读取对内/对外生产知识库绑定(源库 → 目标库)",
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"500": {
"$ref": "#/components/responses/InternalError"
}
}
},
"put": {
"tags": [
"Settings"
],
"operationId": "saveProductionBindings",
"summary": "绑定发布目标的 对内/对外生产知识库(管理员)",
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object"
}
}
}
},
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"500": {
"$ref": "#/components/responses/InternalError"
}
}
}
},
"/api/v1/production-kb-bindings/options": {
"get": {
"tags": [
"Settings"
],
"operationId": "listProductionBindingOptions",
"summary": "列出可绑定的生产 WeKnora 知识库",
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"500": {
"$ref": "#/components/responses/InternalError"
}
}
}
},
"/api/v1/routing-bindings/options": {
"get": {
"tags": [
+12 -8
View File
@@ -101,6 +101,14 @@ func run(ctx context.Context, getenv func(string) string) error {
if err != nil {
return err
}
serviceTargetVerifier := serviceverify.New(serviceverify.Config{
TestAPIKeyFile: config.TestWeKnoraAPIKeyFile,
ProductionAPIKeyFile: config.ProductionWeKnoraAPIKeyFile,
TestBaseURLFallback: config.TestWeKnoraBaseURL,
ProductionBaseURLFallback: config.ProductionWeKnoraBaseURL,
TestAllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP,
ProductionAllowDockerHTTP: config.ProductionWeKnoraAllowDockerHTTP,
})
handler, err := httpapi.NewHandler(httpapi.Config{
Store: postgresstore.NewStore(pool),
Monitor: monitorProvider,
@@ -108,14 +116,10 @@ func run(ctx context.Context, getenv func(string) string) error {
ExternalRoutingKnowledgeBaseID: config.ExternalRoutingKnowledgeBaseID,
RoutingBindings: postgresstore.NewStore(pool),
KnowledgeBases: knowledgeBaseLister,
ServiceTargetVerifier: serviceverify.New(serviceverify.Config{
TestAPIKeyFile: config.TestWeKnoraAPIKeyFile,
ProductionAPIKeyFile: config.ProductionWeKnoraAPIKeyFile,
TestBaseURLFallback: config.TestWeKnoraBaseURL,
ProductionBaseURLFallback: config.ProductionWeKnoraBaseURL,
TestAllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP,
ProductionAllowDockerHTTP: config.ProductionWeKnoraAllowDockerHTTP,
}),
ProductionBindings: postgresstore.NewStore(pool),
ProductionKnowledgeBases: serviceTargetVerifier,
ProductionWeKnoraBaseURL: config.ProductionWeKnoraBaseURL,
ServiceTargetVerifier: serviceTargetVerifier,
ProductionReleaseRequestsEnabled: config.ProductionReleaseRequestsEnabled,
ProductionBackupRestoreVerified: config.ProductionBackupRestoreVerified,
ProductionReleaseAuthorizerRef: config.ProductionReleaseAuthorizerRef,
+18
View File
@@ -0,0 +1,18 @@
# 生产发布的两道门禁(请求门禁 + 执行门禁)。叠加这一份 = 真正「翻闸」。
# 用法:docker compose -f compose.yaml -f compose.release-sim.yaml -f compose.release-gates.yaml up -d api release-worker
# 翻闸后:
# - api 开放「授权发布 / 授权回滚」接口(授权人 = PRODUCTION_RELEASE_AUTHORIZER_REF,Bearer 0600 文件)
# - release-worker 加载发布配置并连接受限的目标数据面
# 关闭:去掉本文件重新 up 即可(门禁默认关闭)。
services:
api:
environment:
PRODUCTION_RELEASE_REQUESTS_ENABLED: "true"
PRODUCTION_BACKUP_RESTORE_VERIFIED: "true"
PRODUCTION_RELEASE_AUTHORIZER_REF: yaozhanghao
PRODUCTION_RELEASE_AUTH_TOKEN_FILE: /run/release-secrets/production_release_auth_token
release-worker:
environment:
RELEASE_WORKER_ENABLED: "true"
RELEASE_DEBUG_ITEM_ERRORS: "true"
+9 -9
View File
@@ -1,21 +1,21 @@
# 本地模拟生产发布专用 override(target=测试服务器 weknora-test-v071)
# 用法:docker compose -f compose.yaml -f compose.release-sim.yaml up -d
# 真实生产严禁使用本文件(真实生产 = 未来独立授权 + 独立配置)
# 发布链路的「数据面接线」override(不含任何门禁开关)
# 用法:docker compose -f compose.yaml -f compose.release-sim.yaml up -d api release-worker
# 叠加顺序:compose.yaml → 本文件(接线)→ compose.release-gates.yaml(门禁,须人工授权)
# 作用:
# 1) api 拿到生产 WeKnora 凭据引用 →「生产知识库绑定」能读真实生产库列表
# 2) release-worker 拿到发布配置、目标端点别名、归档卷(执行门禁关闭时不加载配置)
services:
api:
environment:
PRODUCTION_RELEASE_REQUESTS_ENABLED: "true"
PRODUCTION_BACKUP_RESTORE_VERIFIED: "true"
PRODUCTION_RELEASE_AUTHORIZER_REF: yaozhanghao
PRODUCTION_RELEASE_AUTH_TOKEN_FILE: /run/release-secrets/production_release_auth_token
# 凭据只引用 0600 文件(值不入库、不回显);appconfig 要求两者同时给出
PRODUCTION_WEKNORA_BASE_URL: https://weknora-test.dobot.cn
PRODUCTION_WEKNORA_API_KEY_FILE: /run/release-secrets/target_weknora_api_key
volumes:
- ./secrets/release-sim:/run/release-secrets:ro
release-worker:
environment:
RELEASE_WORKER_ENABLED: "true"
RELEASE_CONFIG_FILE: /run/release-sim/release.sim.json
RELEASE_DEBUG_ITEM_ERRORS: "true"
PRODUCTION_RELEASE_AUTHORIZER_REF: yaozhanghao
extra_hosts:
- "host.docker.internal:host-gateway"
+1 -1
View File
@@ -3,7 +3,7 @@
"routes": [
{
"tenant_id": 10000,
"knowledge_base_id": "c12ff4d5-dd87-423b-ab0f-af745b9d54a7",
"knowledge_base_id": "0c84a098-d3ed-4b82-a165-e6de9691d6a8",
"embedding_model": "text-embedding-v4",
"source": {
"database_url_file": "/run/release-secrets/source_database_url",
@@ -0,0 +1,62 @@
# P8 目标知识库可选(源库 → 目标库映射)开发与验收
## 缺口
发布链路把「测试库」与「生产库」绑成同一个 KB ID,全链路 fail-closed:
- `contract_inspector` 用 `route.knowledge_base_id` 读目标契约,且要求它等于发布项的测试库 ID;
- `manifest.Validate` / `sidecar_publisher` / `verifier` 断言 `KnowledgeBaseID == item.TestKnowledgeBaseID`;
- `postgres_import.prepareSnapshot` 逐行断言 `knowledge_base_id == identity.KnowledgeBaseID`;
- `search_import.searchHitMatches` 断言文档 `knowledge_base_id == identity.KnowledgeBaseID`。
因此只能发布到与测试库同 ID 的「镜像库」,无法选择目标端已有真实库(如「对内知识库-wiki」)。
## 目标
1. 系统设置新增「生产知识库绑定(对内 / 对外)」,从**目标(生产)WeKnora 实际列表**里选目标库,保存时真校验。
2. 发布执行时按绑定把 KB 身份从源库**改写**到目标库,三通道(PostgreSQL / Elasticsearch / 对象)与官方 API 验收一致收敛。
3. 保持 fail-closed 语义:映射关系是**显式声明并被冻结**的,未绑定时退回同 ID 语义(目标库不存在则在契约检查阶段即失败)。
## 契约变更(唯一一处放宽)
原来比较 `manifest.Contract.KnowledgeBaseID == target.KnowledgeBaseID`;
现在比较**声明的映射对**(`releasebundle.KnowledgeBaseMapping`):
```
manifest.Contract.KnowledgeBaseID == mapping.SourceKnowledgeBaseID (源:测试库)
target.KnowledgeBaseID == mapping.TargetKnowledgeBaseID (目标:生产库)
其余字段(版本 / migration / tenant / 检索引擎 / 模型 / 维度 / ES mapping SHA)继续必须相等
```
映射来源:`release_item_executions.target_knowledge_base_id`(执行开始时按 `production_kb_bindings` 解析并落库冻结),审计可见、重试复用。已在生产发布过的条目保留原目标(重试/回滚回到同一处),未发布过的按当前绑定。
## 落点
| 层 | 改动 |
|---|---|
| 迁移 000044 | `production_kb_bindings`(单行 active,对内/对外目标库)+ `release_item_executions.target_knowledge_base_id` |
| 系统设置 | 新增「生产知识库绑定(对内 / 对外)」,从**生产 WeKnora 实际列表**选库;保存时真校验(不存在→400 拒收;生产端不可达→只提示并如实回报 `production_connected=false`) |
| API | `GET/PUT /api/v1/production-kb-bindings`、`GET /api/v1/production-kb-bindings/options` |
| 执行计划 | `StartExecution` 解析映射并冻结进 plan 与 `release_item_executions`;空目标直接拒收 |
| 快照改写 | `DocumentBundle.RemapKnowledgeBase`:5 张带 KB 列的表 + ES 文档 `knowledge_base_id` 改写为目标库,重算表/文档摘要并重新封装(`Seal`);其余字段按原字节透传(嵌入向量不动) |
| 目标身份 | `targetIdentities` / 契约检查器 / 验证器 / `PlanItem.TargetBase()` 全部走映射后的目标库 |
| 对象通道 | 不改:对象 key = `weknora/<tenant>/<knowledge_id>/<asset>`,无 KB 段(实测穷举) |
| Compose | `compose.release-sim.yaml`(数据面接线,无门禁)+ `compose.release-gates.yaml`(两道门禁,须人工授权叠加) |
## 改写面(实测穷举)
源库行里 KB ID 只出现在 `knowledge_base_id` 这一个字段(每行 1 处):
| 表 / 位置 | 是否有 KB 列 | 处理 |
|---|---|---|
| `knowledges` / `chunks` / `knowledge_tags` / `wiki_pages` / `wiki_folders` | 有(各 1 处) | 改写为目标库 ID,并重算表摘要 |
| `resources` / `resource_bindings` / `knowledge_tag_relations` | 无 | 不改(按 knowledge/handle 关联) |
| Elasticsearch `_source.knowledge_base_id` | 有 | 改写为目标库 ID,并重算文档摘要 |
| 对象存储 locator `storage://<backend>/oss://<bucket>/weknora/<tenant>/<knowledge>/<asset>` | 无 KB 段 | 不改(仅按目标 backend/bucket 重写) |
## 验收点
1. `go test ./...`、`go vet`、前端 `tsc` + `vitest` 全绿;迁移 44 up/down 往返可用。
2. 单元/集成:映射生效后 `knowledges/chunks/wiki_pages` 的 KB 列、ES 文档字段均为目标库;摘要随改写重算;未绑定退回同 ID。
3. 界面:系统设置可选目标库(列真实库名),保存时对目标 WeKnora 真校验(无效 ID 拒绝);发布管理显示本次目标库。
4. 真实跨环境:1 份文档发布到目标库 → 三通道读回一致 → 官方 API 检索命中 → 撤权后零命中 → 回滚恢复。
+5
View File
@@ -129,6 +129,11 @@ func classifyRole(method, path string) string {
if path == "/api/v1/routing-bindings" && method != http.MethodGet {
return roleAdmin
}
// Choosing the production base a release writes into is a production
// authority change, exactly like the annotation binding above.
if path == "/api/v1/production-kb-bindings" && method != http.MethodGet {
return roleAdmin
}
// Everything else that mutates (review, batch, seal, confirm, start, retry,
// route visibility, disable) is engineer work; viewers cannot POST at all.
return roleEngineer
+3
View File
@@ -410,6 +410,9 @@ func TestClassifyRoleMatrix(t *testing.T) {
http.MethodPost + " /api/v1/release-executions/rx-1/retry": roleAdmin,
http.MethodPost + " /api/v1/users": roleAdmin,
http.MethodPost + " /api/v1/source-connections/c-1/settings/revisions": roleAdmin,
// Choosing a release's production base is a production authority change.
http.MethodPut + " /api/v1/production-kb-bindings": roleAdmin,
http.MethodGet + " /api/v1/production-kb-bindings": roleViewer,
http.MethodPost + " /api/v1/review-candidates/tp-1/decision": roleEngineer,
http.MethodPost + " /api/v1/release-batches/rb-1/seal": roleEngineer,
http.MethodPost + " /api/v1/test-publications/start": roleEngineer,
+168
View File
@@ -0,0 +1,168 @@
package httpapi
import (
"context"
"net/http"
"strings"
"time"
"knowledge-governance/internal/platform/weknora"
)
// ProductionBindingStore persists the UI-managed 源库 → 目标库 mapping
// (migration 000044). An empty side means "unbound": the release then keeps the
// historical same-id semantics and fails closed at the contract check when no
// such base exists on the target.
type ProductionBindingStore interface {
LoadProductionKBBindings(context.Context) (string, string, string, error)
SaveProductionKBBindings(ctx context.Context, internalID, externalID, updatedBy string, now time.Time) error
}
// ProductionKnowledgeBaseLister enumerates the knowledge bases of the production
// WeKnora, resolved from the operator-managed service configuration.
type ProductionKnowledgeBaseLister interface {
ListProductionKnowledgeBases(ctx context.Context, baseURL string) ([]weknora.KnowledgeBase, error)
}
type productionBindingView struct {
InternalKBID string `json:"internal_kb_id"`
ExternalKBID string `json:"external_kb_id"`
InternalKBName string `json:"internal_kb_name"`
ExternalKBName string `json:"external_kb_name"`
UpdatedBy string `json:"updated_by"`
ProductionBaseURL string `json:"production_base_url"`
ProductionConnected bool `json:"production_connected"`
ProductionMessage string `json:"production_message"`
}
// productionTargetBaseURL is the production endpoint the picker reads from: the
// operator-managed 「服务配置 · 生产地址」 first, the deployment default second.
func (config Config) productionTargetBaseURL(ctx context.Context) string {
if settings, err := config.Store.GetActiveUISettings(ctx); err == nil {
if resolved := strings.TrimSpace(settings.ProductionWeKnoraBaseURL); resolved != "" {
return resolved
}
}
return strings.TrimSpace(config.ProductionWeKnoraBaseURL)
}
// productionKnowledgeBases lists the production-side bases. The second return is
// the honest reason a listing was impossible, so the page can say why instead of
// pretending there are none.
func (config Config) productionKnowledgeBases(ctx context.Context) ([]weknora.KnowledgeBase, string) {
if config.ProductionKnowledgeBases == nil {
return nil, "生产知识库列表未接入(PRODUCTION_WEKNORA_BASE_URL / PRODUCTION_WEKNORA_API_KEY_FILE)"
}
baseURL := config.productionTargetBaseURL(ctx)
if baseURL == "" {
return nil, "未配置生产地址(系统设置 → 服务配置)"
}
bases, err := config.ProductionKnowledgeBases.ListProductionKnowledgeBases(ctx, baseURL)
if err != nil {
return nil, "生产 WeKnora 不可用:" + err.Error()
}
return bases, ""
}
func (config Config) productionBindingPayload(ctx context.Context) (productionBindingView, error) {
storedInternal, storedExternal, updatedBy, err := config.ProductionBindings.LoadProductionKBBindings(ctx)
if err != nil {
return productionBindingView{}, err
}
bases, message := config.productionKnowledgeBases(ctx)
names := make(map[string]string, len(bases))
for _, base := range bases {
names[base.ID] = base.Name
}
return productionBindingView{
InternalKBID: storedInternal, ExternalKBID: storedExternal,
InternalKBName: names[storedInternal], ExternalKBName: names[storedExternal],
UpdatedBy: updatedBy, ProductionBaseURL: config.productionTargetBaseURL(ctx),
ProductionConnected: message == "", ProductionMessage: message,
}, nil
}
func (config Config) handleProductionBindings(w http.ResponseWriter, r *http.Request) {
if config.ProductionBindings == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "production bindings are not configured"})
return
}
view, err := config.productionBindingPayload(r.Context())
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load production bindings"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"data": view})
}
func (config Config) handleProductionKnowledgeBaseOptions(w http.ResponseWriter, r *http.Request) {
bases, message := config.productionKnowledgeBases(r.Context())
if message != "" {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": message})
return
}
options := make([]routingKnowledgeBaseOption, 0, len(bases))
for _, base := range bases {
options = append(options, routingKnowledgeBaseOption{ID: base.ID, Name: base.Name, KnowledgeCount: base.KnowledgeCount})
}
writeJSON(w, http.StatusOK, map[string]any{"data": options})
}
// handleSaveProductionBindings stores the 源库 → 目标库 mapping. When the
// production endpoint answers, the chosen base must exist there; when it cannot
// be asked, the save still succeeds but the response says so honestly (a jittery
// production endpoint must not block the operator, a wrong ID must).
func (config Config) handleSaveProductionBindings(w http.ResponseWriter, r *http.Request) {
if config.ProductionBindings == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "production bindings are not configured"})
return
}
var input struct {
InternalKBID string `json:"internal_kb_id"`
ExternalKBID string `json:"external_kb_id"`
}
if err := decodeStrictJSON(r, &input); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
return
}
internal := strings.TrimSpace(input.InternalKBID)
external := strings.TrimSpace(input.ExternalKBID)
for _, id := range []string{internal, external} {
if id != "" && !validCatalogID(id) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "knowledge base ID is invalid"})
return
}
}
if internal != "" && internal == external {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "对内与对外必须绑定不同的生产知识库"})
return
}
if bases, message := config.productionKnowledgeBases(r.Context()); message == "" {
names := make(map[string]string, len(bases))
for _, base := range bases {
names[base.ID] = base.Name
}
if internal != "" {
if _, ok := names[internal]; !ok {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "对内绑定的生产知识库在该 WeKnora 实例中不存在"})
return
}
}
if external != "" {
if _, ok := names[external]; !ok {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "对外绑定的生产知识库在该 WeKnora 实例中不存在"})
return
}
}
}
if err := config.ProductionBindings.SaveProductionKBBindings(r.Context(), internal, external, actorFor(r), config.Now().UTC()); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not save production bindings"})
return
}
view, err := config.productionBindingPayload(r.Context())
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load production bindings"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"data": view})
}
@@ -0,0 +1,180 @@
package httpapi
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"knowledge-governance/internal/governance"
"knowledge-governance/internal/platform/weknora"
)
type productionBindingStoreFake struct {
internal, external, saver string
savedInternal, savedExternal string
saveCallCount int
}
func (fake *productionBindingStoreFake) LoadProductionKBBindings(context.Context) (string, string, string, error) {
return fake.internal, fake.external, fake.saver, nil
}
func (fake *productionBindingStoreFake) SaveProductionKBBindings(_ context.Context, internal, external, updatedBy string, _ time.Time) error {
fake.savedInternal, fake.savedExternal = internal, external
fake.saver = updatedBy
fake.saveCallCount++
fake.internal, fake.external = internal, external
return nil
}
type productionListerFake struct {
bases []weknora.KnowledgeBase
err error
baseURL string
}
func (fake *productionListerFake) ListProductionKnowledgeBases(_ context.Context, baseURL string) ([]weknora.KnowledgeBase, error) {
fake.baseURL = baseURL
return fake.bases, fake.err
}
func productionBindingsHandler(t *testing.T, store ProductionBindingStore, lister ProductionKnowledgeBaseLister, settingsURL string) http.Handler {
t.Helper()
jobStore := &fakeJobStore{}
jobStore.activeUISettingsResult = governance.UISettings{ProductionWeKnoraBaseURL: settingsURL}
handler, err := NewHandler(Config{
Store: jobStore,
ProductionBindings: store, ProductionKnowledgeBases: lister,
Now: func() time.Time { return time.Date(2026, time.October, 8, 8, 0, 0, 0, time.UTC) },
})
if err != nil {
t.Fatalf("NewHandler: %v", err)
}
return handler
}
func productionBases() []weknora.KnowledgeBase {
return []weknora.KnowledgeBase{
{ID: "kb-prod-internal", Name: "对内知识库-wiki", KnowledgeCount: 2069},
{ID: "kb-prod-external", Name: "对外知识库-wiki", KnowledgeCount: 526},
}
}
// The picker reads the production endpoint from 「服务配置 · 生产地址」 and reports
// the bound bases by NAME — ids never reach the page.
func TestProductionBindingsReadUsesServiceConfiguredEndpoint(t *testing.T) {
store := &productionBindingStoreFake{internal: "kb-prod-internal"}
lister := &productionListerFake{bases: productionBases()}
handler := productionBindingsHandler(t, store, lister, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/production-kb-bindings", nil))
if recorder.Code != http.StatusOK {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
if lister.baseURL != "https://weknora-test.dobot.cn/" {
t.Fatalf("lister baseURL = %q, want the operator-configured production address", lister.baseURL)
}
body := recorder.Body.String()
if !strings.Contains(body, "对内知识库-wiki") || !strings.Contains(body, `"production_connected":true`) {
t.Fatalf("body = %s", body)
}
}
// A chosen base that does not exist on the production endpoint must be refused
// before anything is written.
func TestProductionBindingsSaveRejectsUnknownBase(t *testing.T) {
store := &productionBindingStoreFake{}
handler := productionBindingsHandler(t, store, &productionListerFake{bases: productionBases()}, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPut, "/api/v1/production-kb-bindings",
strings.NewReader(`{"internal_kb_id":"kb-ghost","external_kb_id":""}`)))
if recorder.Code != http.StatusBadRequest {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.saveCallCount != 0 {
t.Fatal("a rejected binding must not be stored")
}
}
func TestProductionBindingsSaveStoresValidatedPair(t *testing.T) {
store := &productionBindingStoreFake{}
handler := productionBindingsHandler(t, store, &productionListerFake{bases: productionBases()}, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPut, "/api/v1/production-kb-bindings",
strings.NewReader(`{"internal_kb_id":"kb-prod-internal","external_kb_id":"kb-prod-external"}`)))
if recorder.Code != http.StatusOK {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.savedInternal != "kb-prod-internal" || store.savedExternal != "kb-prod-external" {
t.Fatalf("saved = %q/%q", store.savedInternal, store.savedExternal)
}
var envelope struct {
Data productionBindingView `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatal(err)
}
if envelope.Data.InternalKBName != "对内知识库-wiki" || !envelope.Data.ProductionConnected {
t.Fatalf("view = %#v", envelope.Data)
}
}
// A jittery production endpoint must not block the operator, but the response
// must say the binding could not be checked.
func TestProductionBindingsSaveWarnsWhenEndpointUnavailable(t *testing.T) {
store := &productionBindingStoreFake{}
handler := productionBindingsHandler(t, store, &productionListerFake{err: errors.New("dial tcp: timeout")}, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPut, "/api/v1/production-kb-bindings",
strings.NewReader(`{"internal_kb_id":"kb-prod-internal","external_kb_id":""}`)))
if recorder.Code != http.StatusOK {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.saveCallCount != 1 {
t.Fatal("an unverifiable (not invalid) binding must still be stored")
}
var envelope struct {
Data productionBindingView `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatal(err)
}
if envelope.Data.ProductionConnected || !strings.Contains(envelope.Data.ProductionMessage, "不可用") {
t.Fatalf("view = %#v", envelope.Data)
}
}
// 对内 and 对外 must not point at the same production base.
func TestProductionBindingsSaveRejectsIdenticalBases(t *testing.T) {
store := &productionBindingStoreFake{}
handler := productionBindingsHandler(t, store, &productionListerFake{bases: productionBases()}, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPut, "/api/v1/production-kb-bindings",
strings.NewReader(`{"internal_kb_id":"kb-prod-internal","external_kb_id":"kb-prod-internal"}`)))
if recorder.Code != http.StatusBadRequest || store.saveCallCount != 0 {
t.Fatalf("status = %d saves = %d", recorder.Code, store.saveCallCount)
}
}
func TestProductionBindingOptionsReportUnavailableReason(t *testing.T) {
handler := productionBindingsHandler(t, &productionBindingStoreFake{}, &productionListerFake{err: errors.New("no key")}, "https://weknora-test.dobot.cn/")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/production-kb-bindings/options", nil))
if recorder.Code != http.StatusBadGateway {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
recorder = httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/production-kb-bindings/options", nil))
// and with a working endpoint the options carry names, not just ids
handler = productionBindingsHandler(t, &productionBindingStoreFake{}, &productionListerFake{bases: productionBases()}, "https://weknora-test.dobot.cn/")
recorder = httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/production-kb-bindings/options", nil))
if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "对外知识库-wiki") {
t.Fatalf("status = %d body %s", recorder.Code, recorder.Body.String())
}
}
+45 -9
View File
@@ -4,7 +4,9 @@ import (
"context"
"net/http"
"strings"
"time"
postgresstore "knowledge-governance/internal/platform/postgres"
"knowledge-governance/internal/platform/weknora"
)
@@ -13,6 +15,9 @@ import (
type RoutingBindingStore interface {
LoadRoutingKBBindings(context.Context) (string, string, string, error)
SaveRoutingKBBindings(context.Context, string, string, string) error
// SaveRoutingKBBindingsWithReplan also re-points 待入库 publications that
// were still waiting for the previous knowledge base (same transaction).
SaveRoutingKBBindingsWithReplan(ctx context.Context, internalID, externalID, effectiveInternal, effectiveExternal, updatedBy string, now time.Time) (postgresstore.RoutingBindingReplanResult, error)
}
// KnowledgeBaseLister enumerates candidate knowledge bases in the connected
@@ -32,6 +37,10 @@ type routingBindingView struct {
ExternalKBName string `json:"external_kb_name"`
UpdatedBy string `json:"updated_by"`
WeKnoraConnected bool `json:"weknora_connected"`
// Replanned / RetiredDuplicates describe what THIS save did to 待入库 items
// that had not been uploaded yet. A plain read reports zeros.
Replanned int `json:"replanned"`
RetiredDuplicates int `json:"retired_duplicates"`
}
type routingKnowledgeBaseOption struct {
@@ -77,21 +86,31 @@ func (config Config) routingBindingNames(ctx context.Context) (map[string]string
return names, true
}
func (config Config) handleRoutingBindings(w http.ResponseWriter, r *http.Request) {
names, connected := config.routingBindingNames(r.Context())
storedInternal, storedExternal, updatedBy, err := config.RoutingBindings.LoadRoutingKBBindings(r.Context())
// routingBindingPayload builds the settings-page projection. replan carries the
// counts of a binding change that just ran (zero on a plain read).
func (config Config) routingBindingPayload(ctx context.Context, replan postgresstore.RoutingBindingReplanResult) (routingBindingView, error) {
names, connected := config.routingBindingNames(ctx)
storedInternal, storedExternal, updatedBy, err := config.RoutingBindings.LoadRoutingKBBindings(ctx)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load routing bindings"})
return
return routingBindingView{}, err
}
internal, external := config.effectiveRoutingTargets(r.Context())
view := routingBindingView{
internal, external := config.effectiveRoutingTargets(ctx)
return routingBindingView{
InternalKBID: internal, ExternalKBID: external,
InternalBoundByUI: storedInternal != "", ExternalBoundByUI: storedExternal != "",
InternalEnvFallback: storedInternal == "" && config.InternalRoutingKnowledgeBaseID != "",
ExternalEnvFallback: storedExternal == "" && config.ExternalRoutingKnowledgeBaseID != "",
InternalKBName: names[storedInternal], ExternalKBName: names[storedExternal],
UpdatedBy: updatedBy, WeKnoraConnected: connected,
Replanned: replan.Replanned, RetiredDuplicates: replan.RetiredDuplicates,
}, nil
}
func (config Config) handleRoutingBindings(w http.ResponseWriter, r *http.Request) {
view, err := config.routingBindingPayload(r.Context(), postgresstore.RoutingBindingReplanResult{})
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load routing bindings"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"data": view})
}
@@ -152,9 +171,26 @@ func (config Config) handleSaveRoutingBindings(w http.ResponseWriter, r *http.Re
}
}
}
if err := config.RoutingBindings.SaveRoutingKBBindings(r.Context(), internal, external, actorFor(r)); err != nil {
// The saved binding re-routes FUTURE annotations; pending 待入库 items that
// have never been uploaded are re-pointed in the same transaction, otherwise
// they would keep aiming at the previous (possibly removed) knowledge base.
effectiveInternal := internal
if effectiveInternal == "" {
effectiveInternal = config.InternalRoutingKnowledgeBaseID
}
effectiveExternal := external
if effectiveExternal == "" {
effectiveExternal = config.ExternalRoutingKnowledgeBaseID
}
replan, err := config.RoutingBindings.SaveRoutingKBBindingsWithReplan(r.Context(), internal, external, effectiveInternal, effectiveExternal, actorFor(r), config.Now().UTC())
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not save routing bindings"})
return
}
config.handleRoutingBindings(w, r)
view, err := config.routingBindingPayload(r.Context(), replan)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load routing bindings"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"data": view})
}
+44
View File
@@ -9,6 +9,7 @@ import (
"testing"
"time"
postgresstore "knowledge-governance/internal/platform/postgres"
"knowledge-governance/internal/platform/weknora"
)
@@ -16,6 +17,11 @@ type bindingStoreFake struct {
internal string
external string
saver string
// replan records what the last save asked for, so tests can assert that the
// effective (env-fallback aware) targets were used.
replanEffective [2]string
replanResult postgresstore.RoutingBindingReplanResult
replanCallCount int
}
func (fake *bindingStoreFake) LoadRoutingKBBindings(context.Context) (string, string, string, error) {
@@ -27,6 +33,13 @@ func (fake *bindingStoreFake) SaveRoutingKBBindings(_ context.Context, internal,
return nil
}
func (fake *bindingStoreFake) SaveRoutingKBBindingsWithReplan(_ context.Context, internal, external, effectiveInternal, effectiveExternal, updatedBy string, _ time.Time) (postgresstore.RoutingBindingReplanResult, error) {
fake.internal, fake.external, fake.saver = internal, external, updatedBy
fake.replanEffective = [2]string{effectiveInternal, effectiveExternal}
fake.replanCallCount++
return fake.replanResult, nil
}
type kbListerFake struct {
bases []weknora.KnowledgeBase
err error
@@ -104,6 +117,37 @@ func TestRoutingBindingsSaveValidatesAgainstWeKnora(t *testing.T) {
}
}
// A binding change must also carry the 待入库 re-plan counts and pass the
// EFFECTIVE targets (env fallback included) so an unbound side still routes.
func TestRoutingBindingsSaveReportsReplanAndUsesEffectiveTargets(t *testing.T) {
store := &bindingStoreFake{replanResult: postgresstore.RoutingBindingReplanResult{Replanned: 3, RetiredDuplicates: 1}}
lister := &kbListerFake{bases: []weknora.KnowledgeBase{{ID: "kb-in", Name: "对内库"}, {ID: "kb-ex", Name: "对外库"}}}
handler := bindingsTestHandler(t, store, lister, "env-internal", "env-external")
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPut, "/api/v1/routing-bindings", strings.NewReader(`{"internal_kb_id":"kb-in","external_kb_id":""}`)))
if recorder.Code != http.StatusOK {
t.Fatalf("save must succeed: %d %s", recorder.Code, recorder.Body.String())
}
if store.replanCallCount != 1 {
t.Fatalf("save must go through the replanning path once, got %d", store.replanCallCount)
}
if store.replanEffective != [2]string{"kb-in", "env-external"} {
t.Fatalf("effective targets = %#v, want the env fallback on the unbound side", store.replanEffective)
}
var envelope struct {
Data struct {
Replanned int `json:"replanned"`
RetiredDuplicates int `json:"retired_duplicates"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatal(err)
}
if envelope.Data.Replanned != 3 || envelope.Data.RetiredDuplicates != 1 {
t.Fatalf("replan counts = %#v", envelope.Data)
}
}
func TestRouteCatalogFilesUsesUIBinding(t *testing.T) {
store := &bindingStoreFake{internal: "ui-internal", external: "ui-external"}
handler := bindingsTestHandler(t, store, &kbListerFake{}, "env-internal", "env-external")
+22
View File
@@ -106,6 +106,11 @@ type Config struct {
ExternalRoutingKnowledgeBaseID string
RoutingBindings RoutingBindingStore
KnowledgeBases KnowledgeBaseLister
// ProductionBindings is the 源库 → 目标库 mapping store; ProductionKnowledgeBases
// enumerates the production-side bases the operator can bind.
ProductionBindings ProductionBindingStore
ProductionKnowledgeBases ProductionKnowledgeBaseLister
ProductionWeKnoraBaseURL string
ServiceTargetVerifier ServiceTargetVerifier
Now func() time.Time
RequireWeKnoraReady bool
@@ -231,6 +236,23 @@ func NewHandler(config Config) (http.Handler, error) {
}
config.handleSaveRoutingBindings(w, r)
})
mux.HandleFunc("GET /api/v1/production-kb-bindings", func(w http.ResponseWriter, r *http.Request) {
if config.ProductionBindings == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "production bindings storage unavailable"})
return
}
config.handleProductionBindings(w, r)
})
mux.HandleFunc("GET /api/v1/production-kb-bindings/options", func(w http.ResponseWriter, r *http.Request) {
config.handleProductionKnowledgeBaseOptions(w, r)
})
mux.HandleFunc("PUT /api/v1/production-kb-bindings", func(w http.ResponseWriter, r *http.Request) {
if config.ProductionBindings == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "production bindings storage unavailable"})
return
}
config.handleSaveProductionBindings(w, r)
})
mux.HandleFunc("POST /api/v1/source-catalog/route", func(w http.ResponseWriter, r *http.Request) {
routeCatalogFiles(w, r, config)
})
@@ -0,0 +1,2 @@
ALTER TABLE release_item_executions DROP COLUMN IF EXISTS target_knowledge_base_id;
DROP TABLE IF EXISTS production_kb_bindings;
@@ -0,0 +1,22 @@
-- Cross-environment release needs an explicit 源库 → 目标库 mapping. Until now
-- the production KB was forced to carry the same id as the tested KB, which
-- meant only same-id "mirror" bases could ever be published to. One row stores
-- which (production-side) knowledge base receives 对内 and which receives 对外
-- releases; an empty side keeps the historical same-id semantics (and fails
-- closed later if that id does not exist on the target).
CREATE TABLE production_kb_bindings (
id TEXT PRIMARY KEY DEFAULT 'active' CHECK (id = 'active'),
internal_kb_id VARCHAR(64) NOT NULL DEFAULT '',
external_kb_id VARCHAR(64) NOT NULL DEFAULT '',
updated_by VARCHAR(64) NOT NULL DEFAULT '',
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
INSERT INTO production_kb_bindings(id,internal_kb_id,external_kb_id,updated_by)
VALUES('active','','','system:migration');
-- The resolved target base is frozen per release item when the execution
-- starts, so an audit shows exactly where each knowledge went and a retry
-- reuses the same target instead of re-resolving a binding that may have moved.
ALTER TABLE release_item_executions
ADD COLUMN target_knowledge_base_id VARCHAR(64) NOT NULL DEFAULT '';
@@ -11,8 +11,8 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
if err != nil {
t.Fatalf("loadMigrations() error = %v", err)
}
if len(migrations) != 43 {
t.Fatalf("migration count = %d, want 43", len(migrations))
if len(migrations) != 44 {
t.Fatalf("migration count = %d, want 44", len(migrations))
}
if migrations[0].Version != 1 || !strings.Contains(migrations[0].UpSQL, "CREATE TABLE IF NOT EXISTS jobs") {
t.Fatalf("migration 1 = %#v", migrations[0])
@@ -0,0 +1,77 @@
package postgres
import (
"context"
"errors"
"time"
"github.com/jackc/pgx/v5"
)
// ProductionBinding is the UI-managed 源库 → 目标库 mapping row. The left side
// of the release (the tested knowledge base) is fixed by the publication; the
// right side is the production-side base this table chooses. An empty side
// keeps the historical same-id semantics.
type ProductionBinding struct {
InternalKBID string
ExternalKBID string
UpdatedBy string
}
// LoadProductionKBBindings reads the single active mapping row. A missing row
// reads back as unbound rather than an error.
func (s *Store) LoadProductionKBBindings(ctx context.Context) (string, string, string, error) {
var binding ProductionBinding
err := s.db.QueryRow(ctx, `SELECT internal_kb_id,external_kb_id,updated_by FROM production_kb_bindings WHERE id='active'`).
Scan(&binding.InternalKBID, &binding.ExternalKBID, &binding.UpdatedBy)
if errors.Is(err, pgx.ErrNoRows) {
return "", "", "", nil
}
if err != nil {
return "", "", "", errors.New("load production kb bindings")
}
return binding.InternalKBID, binding.ExternalKBID, binding.UpdatedBy, nil
}
// SaveProductionKBBindings upserts the mapping row. The knowledge base ids must
// already have been validated against the live production WeKnora by the caller.
func (s *Store) SaveProductionKBBindings(ctx context.Context, internalID, externalID, updatedBy string, now time.Time) error {
if now.IsZero() {
return errors.New("production binding save requires a timestamp")
}
if _, err := s.db.Exec(ctx, `
INSERT INTO production_kb_bindings(id,internal_kb_id,external_kb_id,updated_by,updated_at)
VALUES('active',$1,$2,$3,$4)
ON CONFLICT(id) DO UPDATE SET
internal_kb_id=EXCLUDED.internal_kb_id,
external_kb_id=EXCLUDED.external_kb_id,
updated_by=EXCLUDED.updated_by,
updated_at=EXCLUDED.updated_at`, internalID, externalID, updatedBy, now); err != nil {
return errors.New("save production kb bindings")
}
return nil
}
// ResolveProductionTargetKB maps a publication scope to the production base the
// release should write into. An unbound scope falls back to the tested base id,
// which is the historical same-id behaviour (and fails closed at the contract
// check when no such base exists on the target).
func (s *Store) ResolveProductionTargetKB(ctx context.Context, scope, testKBID string) (string, error) {
internal, external, _, err := s.LoadProductionKBBindings(ctx)
if err != nil {
return "", err
}
bound := ""
switch scope {
case "internal":
bound = internal
case "external":
bound = external
default:
return "", errors.New("unknown release scope")
}
if bound == "" {
return testKBID, nil
}
return bound, nil
}
@@ -0,0 +1,72 @@
package postgres
import (
"context"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// The 源库 → 目标库 mapping has to survive a round trip and, when a scope is
// unbound, fall back to the tested base (the historical same-id semantics).
func TestProductionKBBindingsResolveAndFallBack(t *testing.T) {
databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
if databaseURL == "" {
t.Skip("TEST_DATABASE_URL is not set")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, databaseURL)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
if err = ApplyMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
store := NewStore(pool)
now := time.Date(2026, time.October, 8, 9, 0, 0, 0, time.UTC)
internal, external, _, err := store.LoadProductionKBBindings(ctx)
if err != nil {
t.Fatal(err)
}
if internal != "" || external != "" {
t.Fatalf("fresh mapping = %q/%q, want unbound", internal, external)
}
if target, err := store.ResolveProductionTargetKB(ctx, "internal", "test-kb"); err != nil || target != "test-kb" {
t.Fatalf("unbound resolve = %q/%v, want the tested base", target, err)
}
if err := store.SaveProductionKBBindings(ctx, "prod-internal", "prod-external", "admin", now); err != nil {
t.Fatal(err)
}
if target, err := store.ResolveProductionTargetKB(ctx, "internal", "test-kb"); err != nil || target != "prod-internal" {
t.Fatalf("internal resolve = %q/%v", target, err)
}
if target, err := store.ResolveProductionTargetKB(ctx, "external", "test-kb"); err != nil || target != "prod-external" {
t.Fatalf("external resolve = %q/%v", target, err)
}
if _, err := store.ResolveProductionTargetKB(ctx, "sideways", "test-kb"); err == nil {
t.Fatal("an unknown scope must not resolve to anything")
}
// Unbinding one side returns that side to the same-id semantics only.
if err := store.SaveProductionKBBindings(ctx, "", "prod-external", "admin", now); err != nil {
t.Fatal(err)
}
if target, err := store.ResolveProductionTargetKB(ctx, "internal", "test-kb"); err != nil || target != "test-kb" {
t.Fatalf("unbound side = %q/%v", target, err)
}
if target, _ := store.ResolveProductionTargetKB(ctx, "external", "test-kb"); target != "prod-external" {
t.Fatalf("bound side regressed = %q", target)
}
_, _, saver, err := store.LoadProductionKBBindings(ctx)
if err != nil || saver != "admin" {
t.Fatalf("updated_by = %q (%v)", saver, err)
}
if err := store.SaveProductionKBBindings(ctx, "prod-internal", "", "admin", time.Time{}); err == nil {
t.Fatal("a save without a timestamp must be rejected")
}
}
@@ -63,14 +63,45 @@ execution_gate AS (
AND (gate_version.revision = '' OR gate_publication.parser_generation = '' OR gate_publication.index_generation = '')
)
RETURNING execution.id, execution.release_batch_id, execution.operation, execution.prior_execution_id
), seeded AS (
INSERT INTO release_item_executions (
id, release_execution_id, publication_id, action, state, created_at, updated_at
)
SELECT 'ri-' || SUBSTR(MD5(started.id || ':' || item.publication_id), 1, 32),
started.id, item.publication_id, item.action, 'pending', $4::TIMESTAMPTZ, $4::TIMESTAMPTZ
), target_base AS (
-- 源库 → 目标库 mapping, frozen per item. A publication that was already
-- verified keeps the base it went to (retry / rollback return to the same
-- place); otherwise the UI binding decides, and an unbound scope keeps the
-- historical same-id semantics so existing routings keep working.
SELECT item.publication_id,
COALESCE(
NULLIF(prior_item.target_knowledge_base_id, ''),
NULLIF(CASE publication.target_scope
WHEN 'internal' THEN binding.internal_kb_id
WHEN 'external' THEN binding.external_kb_id
END, ''),
publication.test_kb_id) AS target_kb_id
FROM started
JOIN release_batch_items AS item ON item.release_batch_id = started.release_batch_id
JOIN test_publications AS publication ON publication.id = item.publication_id
LEFT JOIN production_kb_bindings AS binding ON binding.id = 'active'
LEFT JOIN LATERAL (
SELECT candidate.target_knowledge_base_id
FROM release_item_executions AS candidate
JOIN release_executions AS candidate_execution ON candidate_execution.id = candidate.release_execution_id
WHERE candidate.publication_id = item.publication_id
AND candidate.target_knowledge_base_id <> ''
AND candidate.state = 'verified'
AND candidate_execution.operation = 'publish'
AND candidate_execution.state IN ('completed', 'failed')
ORDER BY candidate.completed_at DESC NULLS LAST, candidate.id DESC
LIMIT 1
) AS prior_item ON TRUE
), seeded AS (
INSERT INTO release_item_executions (
id, release_execution_id, publication_id, action, state, created_at, updated_at, target_knowledge_base_id
)
SELECT 'ri-' || SUBSTR(MD5(started.id || ':' || item.publication_id), 1, 32),
started.id, item.publication_id, item.action, 'pending', $4::TIMESTAMPTZ, $4::TIMESTAMPTZ,
COALESCE(target_base.target_kb_id, '')
FROM started
JOIN release_batch_items AS item ON item.release_batch_id = started.release_batch_id
LEFT JOIN target_base ON target_base.publication_id = item.publication_id
WHERE started.operation = 'publish' OR EXISTS (
SELECT 1
FROM release_item_executions AS prior_seed
@@ -97,6 +128,7 @@ execution_gate AS (
publication.parser_generation || '@' || publication.index_generation AS parse_revision,
item.review_id AS release_revision,
publication.test_kb_id,
COALESCE(target_base.target_kb_id, '') AS target_kb_id,
publication.knowledge_id,
item.decided_visibility,
COALESCE(previous.production_revision, '') AS previous_production_revision
@@ -105,6 +137,7 @@ execution_gate AS (
JOIN active_items ON active_items.publication_id = item.publication_id
JOIN test_publications AS publication ON publication.id = item.publication_id
LEFT JOIN target_base ON target_base.publication_id = item.publication_id
JOIN source_item_versions AS current_version ON current_version.id = item.source_item_version_id
LEFT JOIN LATERAL (
SELECT CASE
@@ -146,6 +179,7 @@ SELECT COUNT(DISTINCT started.id)::INTEGER,
'parse_revision', plan.parse_revision,
'release_revision', plan.release_revision,
'test_knowledge_base_id', plan.test_kb_id,
'target_knowledge_base_id', plan.target_kb_id,
'knowledge_id', plan.knowledge_id,
'decided_visibility', plan.decided_visibility,
'previous_production_revision', plan.previous_production_revision
@@ -271,7 +305,7 @@ func (s *Store) StartExecution(ctx context.Context, request releaseworkflow.Star
}
for _, item := range items {
if !validShortID(item.PublicationID) || !validShortID(item.SourceItemVersionID) ||
!validRemoteIdentifier(item.TestKnowledgeBaseID) || !validRemoteIdentifier(item.KnowledgeID) ||
!validRemoteIdentifier(item.TestKnowledgeBaseID) || !validRemoteIdentifier(item.TargetKnowledgeBaseID) || !validRemoteIdentifier(item.KnowledgeID) ||
(item.DecidedVisibility != governance.VisibilityInternal && item.DecidedVisibility != governance.VisibilityExternal) ||
len(item.PreviousProductionRevision) > 128 {
return nil, errors.New("invalid release execution plan")
@@ -13,7 +13,7 @@ import (
)
func TestStartReleaseExecutionSeedsItemsAndReturnsOnlyPendingPlan(t *testing.T) {
payload := `[{"release_execution_id":"rx-safe","publication_id":"tp-safe","action":"delete","source_item_version_id":"sv-safe","source_revision":"source-v1","parse_revision":"parser-v1@index-v1","release_revision":"review-v1","test_knowledge_base_id":"kb-safe","knowledge_id":"knowledge-safe","decided_visibility":"internal","previous_production_revision":"prod-old"}]`
payload := `[{"release_execution_id":"rx-safe","publication_id":"tp-safe","action":"delete","source_item_version_id":"sv-safe","source_revision":"source-v1","parse_revision":"parser-v1@index-v1","release_revision":"review-v1","test_knowledge_base_id":"kb-safe","target_knowledge_base_id":"prod-kb","knowledge_id":"knowledge-safe","decided_visibility":"internal","previous_production_revision":"prod-old"}]`
db := &recordingDB{row: startExecutionRow{executionCount: 1, payload: []byte(payload)}}
items, err := NewStore(db).StartExecution(context.Background(), releaseworkflow.StartExecutionRequest{
ExecutionID: "rx-safe", JobID: "job-safe", Operation: "publish", Now: time.Now(),
@@ -24,6 +24,9 @@ func TestStartReleaseExecutionSeedsItemsAndReturnsOnlyPendingPlan(t *testing.T)
if len(items) != 1 || items[0].PublicationID != "tp-safe" || items[0].Action != "delete" || items[0].KnowledgeID != "knowledge-safe" || items[0].PreviousProductionRevision != "prod-old" {
t.Fatalf("items=%#v", items)
}
if items[0].TargetBase() != "prod-kb" {
t.Fatalf("target base=%q, want the mapped 源库 → 目标库 target", items[0].TargetBase())
}
sql := strings.ToLower(db.queryRowSQL)
for _, required := range []string{
"update release_executions", "then 'rolling_back' else 'running' end", "job_id = $2", "operation = $3",
@@ -34,6 +37,8 @@ func TestStartReleaseExecutionSeedsItemsAndReturnsOnlyPendingPlan(t *testing.T)
"started.operation = 'publish' or exists", "prior_seed.state = 'verified'",
"prior_execution.state in ('completed', 'failed')",
"release_delete_ready", "item.action = 'delete'", "execution_gate.valid_items = execution_gate.item_count",
// the 源库 → 目标库 mapping is resolved into the plan and frozen per item
"production_kb_bindings", "target_knowledge_base_id", "target_base.target_kb_id",
} {
if !strings.Contains(sql, required) {
t.Fatalf("start execution SQL missing %q", required)
@@ -41,6 +46,18 @@ func TestStartReleaseExecutionSeedsItemsAndReturnsOnlyPendingPlan(t *testing.T)
}
}
// An empty resolved target must be rejected rather than silently published to
// an unknown base.
func TestStartReleaseExecutionRejectsEmptyTargetBase(t *testing.T) {
payload := `[{"release_execution_id":"rx-safe","publication_id":"tp-safe","action":"delete","source_item_version_id":"sv-safe","source_revision":"source-v1","parse_revision":"parser-v1@index-v1","release_revision":"review-v1","test_knowledge_base_id":"kb-safe","target_knowledge_base_id":"","knowledge_id":"knowledge-safe","decided_visibility":"internal"}]`
_, err := NewStore(&recordingDB{row: startExecutionRow{executionCount: 1, payload: []byte(payload)}}).StartExecution(context.Background(), releaseworkflow.StartExecutionRequest{
ExecutionID: "rx-safe", JobID: "job-safe", Operation: "publish", Now: time.Now(),
})
if err == nil {
t.Fatal("an unmapped, empty target base must fail closed")
}
}
func TestStartReleaseExecutionRejectsUnknownOrMismatchedExecution(t *testing.T) {
_, err := NewStore(&recordingDB{row: startExecutionRow{executionCount: 0, payload: []byte(`[]`)}}).StartExecution(context.Background(), releaseworkflow.StartExecutionRequest{
ExecutionID: "rx-safe", JobID: "job-safe", Operation: "publish", Now: time.Now(),
@@ -3,6 +3,8 @@ package postgres
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
)
@@ -30,6 +32,17 @@ func (s *Store) LoadRoutingKBBindings(ctx context.Context) (string, string, stri
return binding.InternalKBID, binding.ExternalKBID, binding.UpdatedBy, nil
}
// RoutingBindingReplanResult reports what a binding change did to publications
// that were still waiting in 待入库.
type RoutingBindingReplanResult struct {
// Replanned counts untouched pending publications that were re-pointed at
// the newly bound knowledge base for their scope.
Replanned int
// RetiredDuplicates counts pending publications that already had a twin for
// the new knowledge base, so the stale row was superseded instead of moved.
RetiredDuplicates int
}
// SaveRoutingKBBindings upserts the binding row. Each side is either empty
// (unbind, fall back to env) or a knowledge base id validated by the caller.
func (s *Store) SaveRoutingKBBindings(ctx context.Context, internalID, externalID, updatedBy string) error {
@@ -46,3 +59,98 @@ ON CONFLICT(id) DO UPDATE SET
}
return nil
}
// replanRetireDuplicatesSQL retires a pending publication whose twin for the
// new knowledge base already exists — the unique key is
// (source_item_version_id, test_kb_id), so the row cannot simply be moved.
const replanRetireDuplicatesSQL = `
UPDATE test_publications stale
SET superseded_at = $3, updated_at = $3
WHERE stale.superseded_at IS NULL
AND stale.test_kb_id <> $1
AND stale.processing_state = 'planned'
AND stale.ingest_state = 'pending'
AND stale.knowledge_id = ''
AND stale.retrieval_state = 'unpublished'
AND stale.target_scope = $2
AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type = 'test.upload' AND job.payload->>'publication_id' = stale.id)
AND EXISTS (
SELECT 1 FROM test_publications twin
WHERE twin.source_item_version_id = stale.source_item_version_id
AND twin.test_kb_id = $1
AND twin.superseded_at IS NULL)
`
// replanMovePendingSQL re-points the untouched remainder at the new knowledge
// base. Only rows that have never been uploaded anywhere are eligible: anything
// already in flight keeps its frozen target and stays visible in 待入库.
const replanMovePendingSQL = `
UPDATE test_publications pending
SET test_kb_id = $1, updated_at = $3
WHERE pending.superseded_at IS NULL
AND pending.test_kb_id <> $1
AND pending.processing_state = 'planned'
AND pending.ingest_state = 'pending'
AND pending.knowledge_id = ''
AND pending.retrieval_state = 'unpublished'
AND pending.target_scope = $2
AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type = 'test.upload' AND job.payload->>'publication_id' = pending.id)
AND NOT EXISTS (
SELECT 1 FROM test_publications twin
WHERE twin.source_item_version_id = pending.source_item_version_id
AND twin.test_kb_id = $1
AND twin.superseded_at IS NULL)
`
// SaveRoutingKBBindingsWithReplan saves the binding and, in the SAME
// transaction, re-points 待入库 items that were still waiting for the old
// knowledge base. Without this, changing the binding left every pending
// publication aimed at the previous (possibly deleted) KB, and re-marking the
// file created a SECOND publication instead of moving the first.
//
// effectiveInternal/effectiveExternal are the ids the platform actually routes
// to (the saved binding, or the deployment default when a side is unbound);
// an empty effective id means that scope cannot be routed and is skipped.
func (s *Store) SaveRoutingKBBindingsWithReplan(ctx context.Context, internalID, externalID, effectiveInternal, effectiveExternal, updatedBy string, now time.Time) (RoutingBindingReplanResult, error) {
var result RoutingBindingReplanResult
if now.IsZero() {
return result, errors.New("binding replan requires a timestamp")
}
tx, err := s.db.Begin(ctx)
if err != nil {
return result, fmt.Errorf("begin routing binding save: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `
INSERT INTO routing_kb_bindings(id,internal_kb_id,external_kb_id,updated_by,updated_at)
VALUES('active',$1,$2,$3,$4)
ON CONFLICT(id) DO UPDATE SET
internal_kb_id=EXCLUDED.internal_kb_id,
external_kb_id=EXCLUDED.external_kb_id,
updated_by=EXCLUDED.updated_by,
updated_at=EXCLUDED.updated_at`, internalID, externalID, updatedBy, now); err != nil {
return result, errors.New("save routing kb bindings")
}
for _, scope := range []struct{ name, kbID string }{
{"internal", effectiveInternal},
{"external", effectiveExternal},
} {
if scope.kbID == "" {
continue
}
retired, err := tx.Exec(ctx, replanRetireDuplicatesSQL, scope.kbID, scope.name, now)
if err != nil {
return result, fmt.Errorf("retire duplicate pending publications: %w", err)
}
moved, err := tx.Exec(ctx, replanMovePendingSQL, scope.kbID, scope.name, now)
if err != nil {
return result, fmt.Errorf("replan pending publications: %w", err)
}
result.RetiredDuplicates += int(retired.RowsAffected())
result.Replanned += int(moved.RowsAffected())
}
if err := tx.Commit(ctx); err != nil {
return result, fmt.Errorf("commit routing binding save: %w", err)
}
return result, nil
}
@@ -0,0 +1,115 @@
package postgres
import (
"context"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// Changing the knowledge-base binding used to re-route only FUTURE annotations:
// every publication already sitting in 待入库 kept its frozen test_kb_id, so a
// re-pointed binding left pending items aimed at the previous (deleted) KB, and
// re-marking the file created a SECOND publication instead of moving the first.
func TestSaveRoutingKBBindingsReplansPendingPublications(t *testing.T) {
databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
if databaseURL == "" {
t.Skip("TEST_DATABASE_URL is not set")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, databaseURL)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
if err = ApplyMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
if _, err = pool.Exec(ctx, `
INSERT INTO source_connections(id,source_type,name,enabled) VALUES('conn-replan','manual','Replan',true);
INSERT INTO source_items(id,source_connection_id,stable_external_id,canonical_uri_hash,normalized_path,title,item_type,revision,visibility,visibility_origin) VALUES
('item-in','conn-replan','n-in',repeat('b',64),'docs/in.docx','in.docx','document','locator:r1','internal','configured_scope'),
('item-ext','conn-replan','n-ext',repeat('b',64),'docs/ext.docx','ext.docx','document','locator:r2','external','configured_scope'),
('item-sent','conn-replan','n-sent',repeat('b',64),'docs/sent.docx','sent.docx','document','locator:r3','internal','configured_scope'),
('item-dup','conn-replan','n-dup',repeat('b',64),'docs/dup.docx','dup.docx','document','locator:r4','internal','configured_scope');
INSERT INTO source_item_versions(id,source_item_id,revision,content_object_id,content_state) VALUES
('v-in','item-in','locator:r1',NULL,'unknown'),
('v-ext','item-ext','locator:r2',NULL,'unknown'),
('v-sent','item-sent','locator:r3',NULL,'unknown'),
('v-dup','item-dup','locator:r4',NULL,'unknown');
INSERT INTO test_publications(id,source_item_version_id,test_kb_id,processing_state,ingest_state,target_scope) VALUES
('pub-in','v-in','kb-old-internal','planned','pending','internal'),
('pub-ext','v-ext','kb-old-external','planned','pending','external'),
('pub-sent','v-sent','kb-old-internal','pending','uploaded','internal'),
('pub-dup-old','v-dup','kb-old-internal','planned','pending','internal'),
('pub-dup-new','v-dup','kb-new-internal','planned','pending','internal');
INSERT INTO jobs(id,type,payload,state,idempotency_key) VALUES
('job-sent','test.upload','{"publication_id":"pub-sent"}'::jsonb,'completed','test-start:pub-sent');
`); err != nil {
t.Fatal(err)
}
store := NewStore(pool)
now := time.Date(2026, time.October, 8, 7, 0, 0, 0, time.UTC)
result, err := store.SaveRoutingKBBindingsWithReplan(ctx, "kb-new-internal", "kb-new-external", "kb-new-internal", "kb-new-external", "tester", now)
if err != nil {
t.Fatalf("SaveRoutingKBBindingsWithReplan() error = %v", err)
}
// pub-in and pub-ext move; pub-sent already has an upload job (untouched);
// pub-dup-old already has a twin for the new KB so it is retired instead.
if result.Replanned != 2 || result.RetiredDuplicates != 1 {
t.Fatalf("replan result = %#v, want 2 moved / 1 retired", result)
}
states := map[string]struct {
kb string
superseded bool
}{}
rows, err := pool.Query(ctx, `SELECT id, test_kb_id, superseded_at IS NOT NULL FROM test_publications`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
for rows.Next() {
var id, kb string
var superseded bool
if err := rows.Scan(&id, &kb, &superseded); err != nil {
t.Fatal(err)
}
states[id] = struct {
kb string
superseded bool
}{kb, superseded}
}
for id, want := range map[string]string{
"pub-in": "kb-new-internal", "pub-ext": "kb-new-external", "pub-dup-new": "kb-new-internal",
} {
if states[id].kb != want || states[id].superseded {
t.Fatalf("%s = %#v, want kb %s and not superseded", id, states[id], want)
}
}
if states["pub-sent"].kb != "kb-old-internal" || states["pub-sent"].superseded {
t.Fatalf("an already-uploaded publication must keep its target: %#v", states["pub-sent"])
}
if states["pub-dup-old"].kb != "kb-old-internal" || !states["pub-dup-old"].superseded {
t.Fatalf("the duplicate stale row must be retired: %#v", states["pub-dup-old"])
}
// Re-saving the same binding is a no-op: nothing left to move or retire.
again, err := store.SaveRoutingKBBindingsWithReplan(ctx, "kb-new-internal", "kb-new-external", "kb-new-internal", "kb-new-external", "tester", now)
if err != nil {
t.Fatal(err)
}
if again.Replanned != 0 || again.RetiredDuplicates != 0 {
t.Fatalf("replay must be a no-op, got %#v", again)
}
var internal, external, updatedBy string
if err = pool.QueryRow(ctx, `SELECT internal_kb_id, external_kb_id, updated_by FROM routing_kb_bindings WHERE id='active'`).
Scan(&internal, &external, &updatedBy); err != nil {
t.Fatal(err)
}
if internal != "kb-new-internal" || external != "kb-new-external" || updatedBy != "tester" {
t.Fatalf("binding row = %s / %s / %s", internal, external, updatedBy)
}
}
@@ -64,6 +64,13 @@ func (inspector *TargetContractInspector) Contract(ctx context.Context, item rel
if item.TestKnowledgeBaseID != inspector.config.KnowledgeBaseID {
return releasebundle.TargetContract{}, errors.New("release plan knowledge base does not match target contract")
}
// The route is keyed by the TESTED base (where the content was verified);
// the contract is read for the base this release actually writes into —
// the 源库 → 目标库 mapping when the item carries one, else the same base.
targetKnowledgeBaseID := item.TargetBase()
if !validSearchIdentifier(targetKnowledgeBaseID) {
return releasebundle.TargetContract{}, errors.New("invalid release target knowledge base")
}
var (
migrationVersion int
dirty bool
@@ -73,7 +80,7 @@ func (inspector *TargetContractInspector) Contract(ctx context.Context, item rel
engineType string
indexName string
)
if err := inspector.config.DB.QueryRow(ctx, targetContractSQL, inspector.config.KnowledgeBaseID).Scan(
if err := inspector.config.DB.QueryRow(ctx, targetContractSQL, targetKnowledgeBaseID).Scan(
&migrationVersion, &dirty, &tenantID, &embeddingModel, &dimension, &engineType, &indexName,
); err != nil {
return releasebundle.TargetContract{}, errors.New("read target WeKnora contract")
@@ -92,7 +99,7 @@ func (inspector *TargetContractInspector) Contract(ctx context.Context, item rel
WeKnoraVersion: releasebundle.WeKnoraVersionV071,
MigrationVersion: releasebundle.WeKnoraMigrationV071,
TenantID: tenantID,
KnowledgeBaseID: inspector.config.KnowledgeBaseID,
KnowledgeBaseID: targetKnowledgeBaseID,
RetrieverEngine: "elasticsearch_v8",
EmbeddingModel: embeddingModel,
EmbeddingDimension: dimension,
+34 -16
View File
@@ -70,6 +70,21 @@ func (target *DocumentTarget) Contract(ctx context.Context, item releaseworkflow
return contract, nil
}
// loadBundle loads an archived bundle and applies the item's 源库 → 目标库
// mapping, so activate / compensate / restore / verify all write and read the
// same target base. The archive itself keeps the canonical source-form bundle:
// the remap is deterministic, so a rollback re-derives the same target form.
func (target *DocumentTarget) loadBundle(item releaseworkflow.PlanItem, revision string) (*DocumentBundle, error) {
bundle, err := target.config.Archive.Load(revision)
if err != nil {
return nil, err
}
if err := bundle.RemapKnowledgeBase(item.TargetBase()); err != nil {
return nil, err
}
return bundle, nil
}
func (target *DocumentTarget) Prepare(ctx context.Context, item releaseworkflow.PlanItem, exported releaseworkflow.ExportedBundle) (releaseworkflow.PreparedChange, error) {
bundle, ok := exported.(*DocumentBundle)
if !ok || validateDocumentBundle(bundle) != nil || bundle.Manifest.ReleaseRevision != item.ReleaseRevision {
@@ -93,7 +108,7 @@ func (target *DocumentTarget) Prepare(ctx context.Context, item releaseworkflow.
}
func (target *DocumentTarget) Activate(ctx context.Context, item releaseworkflow.PlanItem, change releaseworkflow.PreparedChange) error {
bundle, err := target.config.Archive.Load(change.ProductionRevision)
bundle, err := target.loadBundle(item, change.ProductionRevision)
if err != nil || bundle.Manifest.ReleaseRevision != item.ReleaseRevision {
return errors.New("load staged v0.7.1 release bundle")
}
@@ -104,7 +119,7 @@ func (target *DocumentTarget) Activate(ctx context.Context, item releaseworkflow
if err := target.config.Objects.Promote(ctx, stage); err != nil {
return errors.New("promote v0.7.1 release assets")
}
postgresIdentity, searchIdentity := targetIdentities(bundle.Manifest)
postgresIdentity, searchIdentity := targetIdentities(item, bundle.Manifest)
if err := target.config.Postgres.Replace(ctx, postgresIdentity, bundle.Postgres); err != nil {
return errors.New("activate v0.7.1 PostgreSQL release")
}
@@ -115,7 +130,7 @@ func (target *DocumentTarget) Activate(ctx context.Context, item releaseworkflow
}
func (target *DocumentTarget) Verify(ctx context.Context, item releaseworkflow.PlanItem, change releaseworkflow.PreparedChange) error {
bundle, err := target.config.Archive.Load(change.ProductionRevision)
bundle, err := target.loadBundle(item, change.ProductionRevision)
if err != nil {
return errors.New("load v0.7.1 release for verification")
}
@@ -130,11 +145,11 @@ func (target *DocumentTarget) Verify(ctx context.Context, item releaseworkflow.P
func (target *DocumentTarget) Compensate(ctx context.Context, item releaseworkflow.PlanItem, change releaseworkflow.PreparedChange) error {
if change.PreviousRevision == "" || change.PreviousRevision == "absent" {
manifestBundle, err := target.config.Archive.Load(change.ProductionRevision)
manifestBundle, err := target.loadBundle(item, change.ProductionRevision)
if err != nil {
return errors.New("load failed first release bundle")
}
postgresIdentity, searchIdentity := targetIdentities(manifestBundle.Manifest)
postgresIdentity, searchIdentity := targetIdentities(item, manifestBundle.Manifest)
if err := target.config.Postgres.Disable(ctx, postgresIdentity); err != nil {
return errors.New("disable failed first PostgreSQL release")
}
@@ -146,16 +161,16 @@ func (target *DocumentTarget) Compensate(ctx context.Context, item releaseworkfl
}
return nil
}
return target.restoreBundle(ctx, change.PreviousRevision)
return target.restoreBundle(ctx, item, change.PreviousRevision)
}
func (target *DocumentTarget) Restore(ctx context.Context, item releaseworkflow.PlanItem, revision string) (string, error) {
if revision == "absent" {
current, err := target.config.Archive.Load(item.ReleaseRevision)
current, err := target.loadBundle(item, item.ReleaseRevision)
if err != nil {
return "", errors.New("load current release for disable")
}
postgresIdentity, searchIdentity := targetIdentities(current.Manifest)
postgresIdentity, searchIdentity := targetIdentities(item, current.Manifest)
if err := target.config.Postgres.Disable(ctx, postgresIdentity); err != nil {
return "", errors.New("disable current PostgreSQL release")
}
@@ -171,7 +186,7 @@ func (target *DocumentTarget) Restore(ctx context.Context, item releaseworkflow.
}
return "absent", nil
}
if err := target.restoreBundle(ctx, revision); err != nil {
if err := target.restoreBundle(ctx, item, revision); err != nil {
return "", err
}
return revision, nil
@@ -182,7 +197,7 @@ func (target *DocumentTarget) VerifyRestored(ctx context.Context, item releasewo
if err := target.config.Verifier.VerifyDisabled(ctx, item); err != nil {
return errors.New("verify disabled production release")
}
current, err := target.config.Archive.Load(item.ReleaseRevision)
current, err := target.loadBundle(item, item.ReleaseRevision)
if err != nil {
return errors.New("load withdrawn v0.7.1 release for wiki verification")
}
@@ -191,7 +206,7 @@ func (target *DocumentTarget) VerifyRestored(ctx context.Context, item releasewo
}
return nil
}
bundle, err := target.config.Archive.Load(revision)
bundle, err := target.loadBundle(item, revision)
if err != nil {
return errors.New("load restored v0.7.1 release")
}
@@ -204,8 +219,8 @@ func (target *DocumentTarget) VerifyRestored(ctx context.Context, item releasewo
return nil
}
func (target *DocumentTarget) restoreBundle(ctx context.Context, revision string) error {
bundle, err := target.config.Archive.Load(revision)
func (target *DocumentTarget) restoreBundle(ctx context.Context, item releaseworkflow.PlanItem, revision string) error {
bundle, err := target.loadBundle(item, revision)
if err != nil {
return errors.New("load prior v0.7.1 release bundle")
}
@@ -216,7 +231,7 @@ func (target *DocumentTarget) restoreBundle(ctx context.Context, revision string
if err := target.config.Objects.Promote(ctx, stage); err != nil {
return errors.New("promote prior v0.7.1 release assets")
}
postgresIdentity, searchIdentity := targetIdentities(bundle.Manifest)
postgresIdentity, searchIdentity := targetIdentities(item, bundle.Manifest)
if err := target.config.Postgres.Replace(ctx, postgresIdentity, bundle.Postgres); err != nil {
return errors.New("restore prior v0.7.1 PostgreSQL release")
}
@@ -226,8 +241,11 @@ func (target *DocumentTarget) restoreBundle(ctx context.Context, revision string
return nil
}
func targetIdentities(manifest releasebundle.Manifest) (PostgresIdentity, SearchIdentity) {
postgresIdentity := PostgresIdentity{TenantID: manifest.Contract.TenantID, KnowledgeBaseID: manifest.Contract.KnowledgeBaseID, KnowledgeID: manifest.Contract.KnowledgeID}
// targetIdentities is where a release item is written: the 源库 → 目标库 mapped
// base (item.TargetBase()), the tenant and the knowledge id that travel with the
// bundle. The bundle's own contract keeps the SOURCE base it was exported from.
func targetIdentities(item releaseworkflow.PlanItem, manifest releasebundle.Manifest) (PostgresIdentity, SearchIdentity) {
postgresIdentity := PostgresIdentity{TenantID: manifest.Contract.TenantID, KnowledgeBaseID: item.TargetBase(), KnowledgeID: manifest.Contract.KnowledgeID}
return postgresIdentity, SearchIdentity(postgresIdentity)
}
+139
View File
@@ -0,0 +1,139 @@
package v071
import (
"encoding/json"
"errors"
"knowledge-governance/internal/releasebundle"
)
// knowledgeBaseField is the single place a knowledge-base identity lives inside
// an exported artefact. Verified by scanning every exported row of a real
// knowledge: the base id appears exactly once per row, in this field.
const knowledgeBaseField = "knowledge_base_id"
// remappableTables are the exported tables whose rows carry that identity.
// resources / resource_bindings / knowledge_tag_relations carry none: they hang
// off the knowledge id, the resource handle and the tag id respectively.
var remappableTables = []string{"knowledges", "chunks", "knowledge_tags", "wiki_pages", "wiki_folders"}
// RemapKnowledgeBase rewrites the bundle's knowledge-base identity from the
// tested base to the target base and re-seals it. Without this, a cross-base
// release would be rejected by the fail-closed guards that assert every row and
// document sits inside the identity scope it is written under.
//
// The manifest keeps the SOURCE base on purpose — that is where the content came
// from, and it is what the release item's tested base is checked against. The
// target base travels with the release item and is asserted separately against
// the target contract.
func (bundle *DocumentBundle) RemapKnowledgeBase(targetKnowledgeBaseID string) error {
if bundle == nil {
return errors.New("release bundle is required")
}
if !validSearchIdentifier(targetKnowledgeBaseID) {
return errors.New("invalid remap target knowledge base")
}
source := bundle.Manifest.Contract.KnowledgeBaseID
if targetKnowledgeBaseID == source {
return nil
}
digests := make(map[string]releasebundle.TableDigest, len(remappableTables))
for _, name := range remappableTables {
table := bundle.Postgres.Table(name)
if len(table.Rows) == 0 {
continue
}
rows := make([]json.RawMessage, 0, len(table.Rows))
for _, row := range table.Rows {
rewritten, err := remapKnowledgeBaseField(row, source, targetKnowledgeBaseID)
if err != nil {
return err
}
rows = append(rows, rewritten)
}
remapped := newTableSnapshot(name, rows)
digests[name] = releasebundle.TableDigest{Name: remapped.Name, RowCount: remapped.RowCount, SHA256: remapped.SHA256}
bundle.Postgres = replaceTable(bundle.Postgres, remapped)
}
searchDigest := bundle.Search.DocumentsSHA256
if len(bundle.Search.Documents) > 0 {
documents := make([]SearchDocument, 0, len(bundle.Search.Documents))
for _, document := range bundle.Search.Documents {
rewritten, err := remapKnowledgeBaseField(document.Source, source, targetKnowledgeBaseID)
if err != nil {
return err
}
documents = append(documents, SearchDocument{ID: document.ID, Source: rewritten})
}
recomputed, err := searchDocumentsDigest(documents)
if err != nil {
return err
}
bundle.Search.Documents = documents
bundle.Search.DocumentsSHA256 = recomputed
searchDigest = recomputed
}
manifest := bundle.Manifest
tables := make([]releasebundle.TableDigest, 0, len(manifest.Tables))
for _, table := range manifest.Tables {
if remapped, ok := digests[table.Name]; ok {
tables = append(tables, remapped)
continue
}
tables = append(tables, table)
}
manifest.Tables = tables
manifest.Search = releasebundle.SearchDigest{DocumentCount: len(bundle.Search.Documents), SHA256: searchDigest}
sealed, err := releasebundle.Seal(manifest)
if err != nil {
return errors.New("reseal remapped release bundle")
}
bundle.Manifest = sealed
return nil
}
// remapKnowledgeBaseField rewrites one JSON object's base identity. Every other
// field is carried through as its original bytes (RawMessage values are never
// re-encoded), so embeddings and other payloads cannot be perturbed by a remap.
func remapKnowledgeBaseField(raw json.RawMessage, source, target string) (json.RawMessage, error) {
var object map[string]json.RawMessage
if err := json.Unmarshal(raw, &object); err != nil {
return nil, errors.New("invalid release row for knowledge base remap")
}
encoded, ok := object[knowledgeBaseField]
if !ok {
return nil, errors.New("release row is missing its knowledge base identity")
}
var current string
if err := json.Unmarshal(encoded, &current); err != nil || current != source {
return nil, errors.New("release row is outside the remapped knowledge base")
}
targetEncoded, err := json.Marshal(target)
if err != nil {
return nil, errors.New("encode remapped knowledge base")
}
object[knowledgeBaseField] = targetEncoded
payload, err := json.Marshal(object)
if err != nil {
return nil, errors.New("encode remapped release row")
}
return json.RawMessage(payload), nil
}
func replaceTable(snapshot PostgresSnapshot, replacement TableSnapshot) PostgresSnapshot {
tables := make([]TableSnapshot, 0, len(snapshot.Tables))
replaced := false
for _, table := range snapshot.Tables {
if table.Name == replacement.Name {
tables = append(tables, replacement)
replaced = true
continue
}
tables = append(tables, table)
}
if !replaced {
tables = append(tables, replacement)
}
return PostgresSnapshot{Tables: tables}
}
+175
View File
@@ -0,0 +1,175 @@
package v071
import (
"encoding/json"
"strings"
"testing"
releaseworkflow "knowledge-governance/internal/workflows/release"
)
// crossBaseBundle is an export fixture whose rows and documents carry the base
// identity the way a real v0.7.1 export does (to_jsonb of every column), which
// is what a 源库 → 目标库 remap has to rewrite.
func crossBaseBundle(t *testing.T) *DocumentBundle {
t.Helper()
pg := PostgresSnapshot{Tables: []TableSnapshot{
newTableSnapshot("knowledges", mustCanonicalRows(t, `{"id":"knowledge-safe","knowledge_base_id":"kb-safe","title":"synthetic release marker"}`)),
newTableSnapshot("chunks", mustCanonicalRows(t,
`{"id":"chunk-safe","knowledge_base_id":"kb-safe","embedding":[0.5,-1,1024]}`,
`{"id":"chunk-two","knowledge_base_id":"kb-safe","embedding":[0.25]}`)),
newTableSnapshot("knowledge_tags", mustCanonicalRows(t, `{"id":"tag-safe","knowledge_base_id":"kb-safe"}`)),
newTableSnapshot("wiki_pages", mustCanonicalRows(t, `{"id":"page-safe","knowledge_base_id":"kb-safe"}`)),
newTableSnapshot("wiki_folders", mustCanonicalRows(t, `{"id":"folder-safe","knowledge_base_id":"kb-safe"}`)),
newTableSnapshot("resources", mustCanonicalRows(t, `{"id":"res-safe","physical_path":"storage://backend/oss://bucket/weknora/10000/knowledge-safe/asset"}`)),
}}
search := SearchSnapshot{Documents: []SearchDocument{{
ID: "es-safe",
Source: mustCanonicalRows(t, `{"tenant_id":10000,"knowledge_base_id":"kb-safe","knowledge_id":"knowledge-safe","embedding":[0.5,-1,1024]}`)[0],
}}}
search.DocumentsSHA256, _ = searchDocumentsDigest(search.Documents)
search.MappingSHA256 = strings.Repeat("a", 64)
exporter, err := NewDocumentExporter(DocumentExporterConfig{
Contract: &stubSourceContract{contract: sourceContractFixture(search.MappingSHA256)},
Postgres: &stubPGExporter{snapshot: pg}, Search: &stubSearchExporter{snapshot: search},
Objects: &stubObjectExporter{snapshot: ObjectSnapshot{SHA256: objectSnapshotDigest(nil)}},
TenantID: 10000, EmbeddingModel: "bge-m3", Now: fixedArchiveTime,
})
if err != nil {
t.Fatal(err)
}
value, err := exporter.Export(t.Context(), releasePlanForArchive())
if err != nil {
t.Fatal(err)
}
return value.(*DocumentBundle)
}
func rowField(t *testing.T, row json.RawMessage, field string) string {
t.Helper()
var object map[string]json.RawMessage
if err := json.Unmarshal(row, &object); err != nil {
t.Fatal(err)
}
var value string
if err := json.Unmarshal(object[field], &value); err != nil {
t.Fatal(err)
}
return value
}
// The whole point of the production binding: a knowledge verified in one base
// can be written into a DIFFERENT one, with every row and document moved over
// and the bundle re-sealed so the fail-closed checks still hold.
func TestRemapKnowledgeBaseRewritesAllChannelsAndReseals(t *testing.T) {
bundle := crossBaseBundle(t)
sourceBundleSHA := bundle.Manifest.BundleSHA256
chunkBefore := string(bundle.Postgres.Table("chunks").Rows[0])
resourceBefore := string(bundle.Postgres.Table("resources").Rows[0])
if err := bundle.RemapKnowledgeBase("prod-kb"); err != nil {
t.Fatalf("remap: %v", err)
}
// The manifest keeps the SOURCE base: that is where the content came from and
// what the release item's tested base is checked against.
if got := bundle.Manifest.Contract.KnowledgeBaseID; got != "kb-safe" {
t.Fatalf("manifest base = %q, want the source base", got)
}
for _, name := range remappableTables {
table := bundle.Postgres.Table(name)
if len(table.Rows) == 0 {
t.Fatalf("table %s lost its rows", name)
}
for _, row := range table.Rows {
if got := rowField(t, row, knowledgeBaseField); got != "prod-kb" {
t.Fatalf("%s row base = %q, want prod-kb", name, got)
}
if strings.Contains(string(row), "kb-safe") {
t.Fatalf("%s row still mentions the source base: %s", name, row)
}
}
}
// Tables without a base column are carried through untouched.
if got := string(bundle.Postgres.Table("resources").Rows[0]); got != resourceBefore {
t.Fatalf("resources row changed: %s", got)
}
// Embeddings must survive byte-for-byte: only the identity field is rewritten.
var before, after map[string]json.RawMessage
if err := json.Unmarshal([]byte(chunkBefore), &before); err != nil {
t.Fatal(err)
}
if err := json.Unmarshal(bundle.Postgres.Table("chunks").Rows[0], &after); err != nil {
t.Fatal(err)
}
if string(before["embedding"]) != string(after["embedding"]) {
t.Fatalf("embedding payload was re-encoded: %s -> %s", before["embedding"], after["embedding"])
}
document := bundle.Search.Documents[0]
if got := rowField(t, document.Source, knowledgeBaseField); got != "prod-kb" {
t.Fatalf("document base = %q, want prod-kb", got)
}
// Every digest the target verifies against is recomputed and re-sealed.
if err := validateDocumentBundle(bundle); err != nil {
t.Fatalf("remapped bundle must stay self-consistent: %v", err)
}
recomputed, err := searchDocumentsDigest(bundle.Search.Documents)
if err != nil || recomputed != bundle.Search.DocumentsSHA256 || recomputed != bundle.Manifest.Search.SHA256 {
t.Fatalf("search digest not coherent: %v %s %s", err, recomputed, bundle.Manifest.Search.SHA256)
}
if bundle.Manifest.BundleSHA256 == sourceBundleSHA {
t.Fatal("bundle digest must change when the content identity changes")
}
}
// A same-base release (the historical semantics) must not touch the bundle at all.
func TestRemapKnowledgeBaseIsNoOpForSameBase(t *testing.T) {
bundle := crossBaseBundle(t)
before := bundle.Manifest.BundleSHA256
if err := bundle.RemapKnowledgeBase("kb-safe"); err != nil {
t.Fatal(err)
}
if bundle.Manifest.BundleSHA256 != before {
t.Fatalf("same-base remap changed the bundle digest")
}
}
func TestRemapKnowledgeBaseFailsClosed(t *testing.T) {
tests := []struct {
name string
mutate func(*DocumentBundle)
target string
}{
{name: "invalid target", target: "prod/kb", mutate: func(*DocumentBundle) {}},
{name: "empty target", target: "", mutate: func(*DocumentBundle) {}},
{name: "row outside the source base", target: "prod-kb", mutate: func(bundle *DocumentBundle) {
bundle.Postgres = replaceTable(bundle.Postgres, newTableSnapshot("chunks", mustCanonicalRows(t, `{"id":"chunk-safe","knowledge_base_id":"other-kb"}`)))
}},
{name: "row without an identity", target: "prod-kb", mutate: func(bundle *DocumentBundle) {
bundle.Postgres = replaceTable(bundle.Postgres, newTableSnapshot("knowledges", mustCanonicalRows(t, `{"id":"knowledge-safe"}`)))
}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
bundle := crossBaseBundle(t)
tt.mutate(bundle)
if err := bundle.RemapKnowledgeBase(tt.target); err == nil {
t.Fatal("RemapKnowledgeBase() error=nil")
}
})
}
if err := (*DocumentBundle)(nil).RemapKnowledgeBase("prod-kb"); err == nil {
t.Fatal("nil bundle must be rejected")
}
}
// TargetBase is what every target-side write keys on.
func TestPlanItemTargetBaseFallsBackToTestedBase(t *testing.T) {
mapped := releaseworkflow.PlanItem{TestKnowledgeBaseID: "kb-safe", TargetKnowledgeBaseID: "prod-kb"}
if mapped.TargetBase() != "prod-kb" {
t.Fatalf("mapped TargetBase = %q", mapped.TargetBase())
}
unbound := releaseworkflow.PlanItem{TestKnowledgeBaseID: "kb-safe"}
if unbound.TargetBase() != "kb-safe" {
t.Fatalf("unbound TargetBase = %q, want the tested base", unbound.TargetBase())
}
}
+2 -2
View File
@@ -34,7 +34,7 @@ func (verifier *WeKnoraVerifier) Verify(ctx context.Context, item releaseworkflo
return errors.New("invalid WeKnora verification bundle")
}
knowledge, err := verifier.client.GetKnowledge(ctx, item.KnowledgeID)
if err != nil || knowledge.ID != item.KnowledgeID || knowledge.KnowledgeBaseID != item.TestKnowledgeBaseID || knowledge.ParseStatus != "completed" {
if err != nil || knowledge.ID != item.KnowledgeID || knowledge.KnowledgeBaseID != item.TargetBase() || knowledge.ParseStatus != "completed" {
return errors.New("production WeKnora knowledge verification failed")
}
_, chunkTotal, err := verifier.client.ListChunks(ctx, item.KnowledgeID, 1, 1, bundleChunkTypes(bundle))
@@ -167,7 +167,7 @@ func (verifier *WeKnoraVerifier) VerifyWiki(ctx context.Context, item releasewor
seen := map[string]bool{}
scanned := 0
for page := 1; ; page++ {
pages, total, err := verifier.client.ListWikiPages(ctx, item.TestKnowledgeBaseID, page, pageSize)
pages, total, err := verifier.client.ListWikiPages(ctx, item.TargetBase(), page, pageSize)
if err != nil {
return errors.New("production WeKnora wiki verification failed")
}
+28 -2
View File
@@ -99,7 +99,27 @@ func Seal(manifest Manifest) (Manifest, error) {
return manifest, nil
}
func Validate(manifest Manifest, target TargetContract) error {
// KnowledgeBaseMapping is the declared 源库 → 目标库 pair for a release item.
// An empty TargetKnowledgeBaseID means the historical same-id semantics: the
// production base must carry exactly the tested base's id.
type KnowledgeBaseMapping struct {
SourceKnowledgeBaseID string
TargetKnowledgeBaseID string
}
// EffectiveTarget is the base the release writes into.
func (mapping KnowledgeBaseMapping) EffectiveTarget() string {
if trimmed := strings.TrimSpace(mapping.TargetKnowledgeBaseID); trimmed != "" {
return trimmed
}
return mapping.SourceKnowledgeBaseID
}
// Validate proves the bundle contract against the target data plane. The only
// thing a cross-base release relaxes is the base identity itself: the bundle
// keeps the SOURCE base and the target must match the DECLARED mapping, so an
// undeclared (or stale) mapping still fails closed.
func Validate(manifest Manifest, target TargetContract, mapping KnowledgeBaseMapping) error {
if err := validateManifestFields(manifest); err != nil {
return err
}
@@ -114,9 +134,15 @@ func Validate(manifest Manifest, target TargetContract) error {
return errors.New("bundle digest mismatch")
}
contract := manifest.Contract
if mapping.SourceKnowledgeBaseID != contract.KnowledgeBaseID {
return errors.New("release mapping does not match bundle")
}
if target.KnowledgeBaseID != mapping.EffectiveTarget() {
return errors.New("target release contract mismatch")
}
if target.WeKnoraVersion != WeKnoraVersionV071 || target.MigrationVersion != WeKnoraMigrationV071 ||
contract.WeKnoraVersion != target.WeKnoraVersion || contract.MigrationVersion != target.MigrationVersion ||
contract.TenantID != target.TenantID || contract.KnowledgeBaseID != target.KnowledgeBaseID ||
contract.TenantID != target.TenantID ||
contract.RetrieverEngine != target.RetrieverEngine || contract.EmbeddingModel != target.EmbeddingModel ||
contract.EmbeddingDimension != target.EmbeddingDimension ||
contract.ElasticsearchMappingSHA256 != target.ElasticsearchMappingSHA256 {
+50 -2
View File
@@ -40,7 +40,7 @@ func TestSealAndValidateV071Manifest(t *testing.T) {
EmbeddingModel: "bge-m3", EmbeddingDimension: 1024,
ElasticsearchMappingSHA256: strings.Repeat("a", 64),
}
if err := Validate(sealed, target); err != nil {
if err := Validate(sealed, target, sameIDMapping()); err != nil {
t.Fatal(err)
}
encoded, err := json.Marshal(sealed)
@@ -96,7 +96,7 @@ func TestValidateRejectsContractDriftAndUnknownTables(t *testing.T) {
candidate, contract := manifest, target
candidate.Tables = append([]TableDigest(nil), manifest.Tables...)
tt.mutate(&candidate, &contract)
if err := Validate(candidate, contract); err == nil {
if err := Validate(candidate, contract, sameIDMapping()); err == nil {
t.Fatal("Validate() error=nil")
}
})
@@ -117,3 +117,51 @@ func validManifest() Manifest {
func validTarget() TargetContract {
return TargetContract{WeKnoraVersion: "v0.7.1", MigrationVersion: 74, TenantID: 10000, KnowledgeBaseID: "kb-safe", RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "bge-m3", EmbeddingDimension: 1024, ElasticsearchMappingSHA256: strings.Repeat("a", 64)}
}
// sameIDMapping is the historical semantics: no declared target, so the
// production base must carry the tested base's id.
func sameIDMapping() KnowledgeBaseMapping {
return KnowledgeBaseMapping{SourceKnowledgeBaseID: "kb-safe"}
}
// A declared 源库 → 目标库 mapping lets the bundle land in a DIFFERENT base,
// which is the whole point of the production binding setting.
func TestValidateAcceptsDeclaredCrossBaseMapping(t *testing.T) {
manifest, err := Seal(validManifest())
if err != nil {
t.Fatal(err)
}
target := validTarget()
target.KnowledgeBaseID = "prod-kb"
mapping := KnowledgeBaseMapping{SourceKnowledgeBaseID: "kb-safe", TargetKnowledgeBaseID: "prod-kb"}
if err := Validate(manifest, target, mapping); err != nil {
t.Fatalf("declared cross-base release must validate: %v", err)
}
}
// An undeclared, stale or self-inconsistent mapping must fail closed: neither
// "target drifted somewhere else" nor "mapping points at the source" passes.
func TestValidateRejectsUndeclaredOrDriftedMapping(t *testing.T) {
manifest, err := Seal(validManifest())
if err != nil {
t.Fatal(err)
}
tests := []struct {
name string
mapping KnowledgeBaseMapping
target string
}{
{name: "mapping source does not match bundle", mapping: KnowledgeBaseMapping{SourceKnowledgeBaseID: "other"}, target: "other"},
{name: "target differs from declared mapping", mapping: KnowledgeBaseMapping{SourceKnowledgeBaseID: "kb-safe", TargetKnowledgeBaseID: "prod-kb"}, target: "another-kb"},
{name: "mapping target differs while target keeps source id", mapping: KnowledgeBaseMapping{SourceKnowledgeBaseID: "kb-safe", TargetKnowledgeBaseID: "prod-kb"}, target: "kb-safe"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
target := validTarget()
target.KnowledgeBaseID = tt.target
if err := Validate(manifest, target, tt.mapping); err == nil {
t.Fatal("Validate() error=nil")
}
})
}
}
@@ -0,0 +1,29 @@
package serviceverify
import (
"context"
"errors"
"knowledge-governance/internal/platform/weknora"
)
// ListProductionKnowledgeBases enumerates the bases a release could be written
// into. It asks the operator-managed production endpoint with the production
// credential file, so the settings page can offer real bases instead of asking
// the operator to paste an id. A missing endpoint or credential is an error the
// caller reports honestly rather than a silently empty picker.
func (verifier *Verifier) ListProductionKnowledgeBases(ctx context.Context, baseURL string) ([]weknora.KnowledgeBase, error) {
resolved := firstNonEmpty(baseURL, verifier.config.ProductionBaseURLFallback)
if resolved == "" {
return nil, errors.New("production WeKnora endpoint is not configured")
}
apiKey := verifier.apiKey(verifier.config.ProductionAPIKeyFile)
if apiKey == "" {
return nil, errors.New("production WeKnora API key file is not available")
}
client, err := verifier.client(resolved, apiKey, verifier.config.ProductionAllowDockerHTTP)
if err != nil {
return nil, err
}
return client.ListKnowledgeBases(ctx)
}
+14
View File
@@ -45,6 +45,10 @@ type PlanItem struct {
ParseRevision string `json:"parse_revision"`
ReleaseRevision string `json:"release_revision"`
TestKnowledgeBaseID string `json:"test_knowledge_base_id"`
// TargetKnowledgeBaseID is the 源库 → 目标库 mapping resolved when the
// execution started (see production_kb_bindings). It equals the tested base
// when the scope is unbound, which is the historical same-id semantics.
TargetKnowledgeBaseID string `json:"target_knowledge_base_id"`
KnowledgeID string `json:"knowledge_id"`
DecidedVisibility governance.Visibility `json:"decided_visibility"`
PreviousProductionRevision string `json:"previous_production_revision"`
@@ -56,6 +60,16 @@ type ItemResult struct {
ProductionRevision string
}
// TargetBase is the knowledge base this item is written to: the resolved
// 源库 → 目标库 mapping when the release carries one, otherwise the tested base
// itself (the historical same-id semantics).
func (item PlanItem) TargetBase() string {
if trimmed := strings.TrimSpace(item.TargetKnowledgeBaseID); trimmed != "" {
return trimmed
}
return item.TestKnowledgeBaseID
}
type StartExecutionRequest struct {
ExecutionID string
JobID string
@@ -43,6 +43,15 @@ func NewSidecarPublisher(exporter BundleExporter, target SidecarTarget) (*Sideca
return &SidecarPublisher{exporter: exporter, target: target}, nil
}
// releaseItemMapping declares which production base this item must land in: the
// tested base it came from, and the 源库 → 目标库 target the execution resolved.
func releaseItemMapping(item PlanItem) releasebundle.KnowledgeBaseMapping {
return releasebundle.KnowledgeBaseMapping{
SourceKnowledgeBaseID: item.TestKnowledgeBaseID,
TargetKnowledgeBaseID: item.TargetKnowledgeBaseID,
}
}
func (publisher *SidecarPublisher) Publish(ctx context.Context, item PlanItem) (ItemResult, error) {
bundle, err := publisher.exporter.Export(ctx, item)
if err != nil {
@@ -62,7 +71,7 @@ func (publisher *SidecarPublisher) Publish(ctx context.Context, item PlanItem) (
if err != nil {
return ItemResult{}, newSidecarError(ErrorCodeTargetUnavailable, true, err)
}
if err := releasebundle.Validate(manifest, targetContract); err != nil {
if err := releasebundle.Validate(manifest, targetContract, releaseItemMapping(item)); err != nil {
return ItemResult{}, newSidecarError(ErrorCodeContractMismatch, false, err)
}
change, err := publisher.target.Prepare(ctx, item, bundle)
@@ -121,7 +130,7 @@ func (publisher *SidecarPublisher) Delete(ctx context.Context, item PlanItem) (I
if err != nil {
return ItemResult{}, newSidecarError(ErrorCodeTargetUnavailable, true, err)
}
if err := releasebundle.Validate(manifest, targetContract); err != nil {
if err := releasebundle.Validate(manifest, targetContract, releaseItemMapping(item)); err != nil {
return ItemResult{}, newSidecarError(ErrorCodeContractMismatch, false, err)
}
change, err := publisher.target.Prepare(ctx, item, bundle)
+3 -1
View File
@@ -16,7 +16,9 @@ echo "=== 3. 构建 web"
docker build --network host "${PROXY_ARGS[@]}" -f Dockerfile.web -t knowledge-governance-p1-web:latest . || { echo "WEB_BUILD_FAILED"; exit 1; }
echo "=== 4. 上线 api + web(不动依赖)"
docker compose up -d --no-build --no-deps api web || { echo "UP_FAILED"; exit 1; }
# COMPOSE_FILES 让调用方叠加发布接线 / 门禁 override,默认只有基础 compose.yaml。
# 例:COMPOSE_FILES="-f compose.yaml -f compose.release-sim.yaml" bash scripts/deploy-api-web.sh
docker compose ${COMPOSE_FILES:--f compose.yaml} up -d --no-build --no-deps api web || { echo "UP_FAILED"; exit 1; }
echo "=== 5. 状态"
docker ps --filter name=knowledge-governance-p1 --format '{{.Names}} {{.Status}} {{.Image}}'
+87
View File
@@ -0,0 +1,87 @@
import { useCallback, useEffect, useState } from 'react'
import { useOptionalAuth } from './auth'
import { fetchProductionBindingOptions, fetchProductionBindings, saveProductionBindings } from './productionBindings'
import type { ProductionBindings } from './productionBindings'
import type { KnowledgeBaseOption } from './routingBindings'
// 「生产知识库绑定」:发布链路的 源库 → 目标库 映射。这里选的是发布要写入的
// 生产(目标)知识库,与上面的「知识库绑定」(测试端收件库)是两件事。
export function ProductionBindingsSection() {
const auth = useOptionalAuth()
const isAdmin = auth?.identity?.role === 'admin'
const [bindings, setBindings] = useState<ProductionBindings | null>(null)
const [options, setOptions] = useState<KnowledgeBaseOption[]>([])
const [internalDraft, setInternalDraft] = useState('')
const [externalDraft, setExternalDraft] = useState('')
const [loading, setLoading] = useState(true)
const [saving, setSaving] = useState(false)
const [error, setError] = useState<string | null>(null)
const [message, setMessage] = useState('')
const load = useCallback(async () => {
setLoading(true); setError(null)
try {
const next = await fetchProductionBindings()
setBindings(next)
setInternalDraft(next.internal_kb_id)
setExternalDraft(next.external_kb_id)
try { setOptions(await fetchProductionBindingOptions()) } catch { setOptions([]) }
} catch (cause) { setError(cause instanceof Error ? cause.message : '读取生产知识库绑定失败') }
finally { setLoading(false) }
}, [])
useEffect(() => { void load() }, [load])
const dirty = bindings !== null && (internalDraft !== bindings.internal_kb_id || externalDraft !== bindings.external_kb_id)
const sameSelection = internalDraft !== '' && internalDraft === externalDraft
const save = async () => {
setSaving(true); setError(null); setMessage('')
try {
const next = await saveProductionBindings({ internal_kb_id: internalDraft, external_kb_id: externalDraft })
setBindings(next)
const where = `对内 ${next.internal_kb_name || next.internal_kb_id.slice(0, 8) || '未绑定'} · 对外 ${next.external_kb_name || next.external_kb_id.slice(0, 8) || '未绑定'}`
setMessage(next.production_connected
? `发布目标已更新:${where}。之后新发布的“对内/对外”内容会写入这两个生产库(已发布过的条目仍回到原目标)。`
: `发布目标已保存:${where};但当前核对不了生产端(${next.production_message}),下次发布前请确认。`)
} catch (cause) { setError(cause instanceof Error ? cause.message : '保存生产知识库绑定失败') }
finally { setSaving(false) }
}
const optionLabel = (base: KnowledgeBaseOption) => `${base.name}(${base.knowledge_count} 文档)`
const renderSide = (label: string, boundID: string, boundName: string) => (
<div className={`kb-binding-side ${boundID ? 'bound' : 'warn'}`}>
<b>{label}</b>
<span>{boundID ? (boundName || `${boundID.slice(0, 8)}…`) : '未绑定 · 按同 ID 语义发布(目标端须存在同 ID 库)'}</span>
<small>{boundID ? 'UI 绑定 · 发布写入此库' : '未绑定:沿用测试库 ID'}</small>
</div>
)
const editorFor = (value: string, onChange: (next: string) => void, placeholder: string) => (
options.length > 0
? <select value={value} onChange={(event) => onChange(event.target.value)} aria-label={placeholder}>
<option value="">{placeholder}</option>
{options.map((base) => <option key={base.id} value={base.id}>{optionLabel(base)}</option>)}
</select>
: <input value={value} onChange={(event) => onChange(event.target.value.trim())} placeholder={`${placeholder}(输入生产知识库 ID)`} aria-label={placeholder} />
)
return <section className="settings-panel wide" id="settings-production-bindings">
<div className="settings-section-heading"><div><h2>生产知识库绑定</h2><p>发布链路的“源库 → 目标库”:新发布的对内 / 对外内容按这里的绑定写入生产 WeKnora 的对应知识库。留空表示未绑定,按同 ID 语义发布。</p></div><button onClick={() => void load()} disabled={loading}>{loading ? '读取中…' : '重新读取'}</button></div>
{error ? <p className="error-message" role="alert">{error}</p> : null}
{bindings ? <div className="kb-binding-grid">
{renderSide('对内生产知识库', bindings.internal_kb_id, bindings.internal_kb_name)}
{renderSide('对外生产知识库', bindings.external_kb_id, bindings.external_kb_name)}
</div> : null}
{bindings && !bindings.production_connected ? <p className="hint">生产端核对不可用:{bindings.production_message}。{bindings.production_base_url ? `(生产地址:${bindings.production_base_url})` : ''}</p> : null}
{bindings && bindings.production_connected && bindings.production_base_url ? <p className="hint">生产端:{bindings.production_base_url}{bindings.updated_by ? ` · 最近修改:${bindings.updated_by}` : ''}</p> : null}
{isAdmin && bindings ? <div className="kb-binding-editor">
<label>对内目标库 {editorFor(internalDraft, setInternalDraft, '选择对内生产知识库')}</label>
<label>对外目标库 {editorFor(externalDraft, setExternalDraft, '选择对外生产知识库')}</label>
<p className="hint">保存时会向生产端核对所选的库是否真实存在;对不上会拒收,生产端不可达则只提示。</p>
{sameSelection ? <p className="error-message">对内与对外必须绑定不同的生产知识库</p> : null}
<button onClick={() => void save()} disabled={saving || !dirty || sameSelection}>{saving ? '正在核对并保存…' : '保存发布目标'}</button>
{message ? <p className="success-message" role="status">{message}</p> : null}
</div> : null}
{!isAdmin ? <p className="hint">仅管理员可修改发布目标。</p> : null}
</section>
}
+4 -1
View File
@@ -37,7 +37,10 @@ export function RoutingBindingsSection() {
try {
const next = await saveRoutingBindings({ internal_kb_id: internalDraft, external_kb_id: externalDraft })
setBindings(next)
setMessage(`绑定已生效:对内 ${next.internal_kb_name || next.internal_kb_id.slice(0, 8) || '未绑定'} · 对外 ${next.external_kb_name || next.external_kb_id.slice(0, 8) || '未绑定'}。此后“来源与目录”的标记按新绑定路由。`)
const replan = next.replanned || next.retired_duplicates
? `;同时把 ${next.replanned} 项还没上传的待入库重新指向新库${next.retired_duplicates ? `,${next.retired_duplicates} 项因新库里已有同篇而被退休` : ''}`
: ''
setMessage(`绑定已生效:对内 ${next.internal_kb_name || next.internal_kb_id.slice(0, 8) || '未绑定'} · 对外 ${next.external_kb_name || next.external_kb_id.slice(0, 8) || '未绑定'}。此后“来源与目录”的标记按新绑定路由${replan}。`)
} catch (cause) { setError(cause instanceof Error ? cause.message : '保存知识库绑定失败') }
finally { setSaving(false) }
}
+4 -1
View File
@@ -3,6 +3,7 @@ import { useCallback, useEffect, useState } from 'react'
import { MembersSection } from './MembersSection'
import { DingtalkSessionControl } from './DingtalkSessionControl'
import { RoutingBindingsSection } from './RoutingBindingsSection'
import { ProductionBindingsSection } from './ProductionBindingsSection'
import { DingtalkSpaceRegistrar } from './DingtalkSpaceRegistrar'
import { FilenameFilterEditor } from './FilenameFilterEditor'
import { useOptionalAuth } from './auth'
@@ -35,12 +36,13 @@ interface Props {
initialSection?: SectionKey
}
type SectionKey = 'sources' | 'services' | 'bindings' | 'storage' | 'refresh' | 'members'
type SectionKey = 'sources' | 'services' | 'bindings' | 'production-bindings' | 'storage' | 'refresh' | 'members'
const sections: { key: SectionKey; label: string; adminOnly?: boolean }[] = [
{ key: 'sources', label: '数据源配置' },
{ key: 'services', label: '服务配置' },
{ key: 'bindings', label: '知识库绑定' },
{ key: 'production-bindings', label: '生产知识库绑定' },
{ key: 'storage', label: '存储与备份' },
{ key: 'refresh', label: '刷新策略' },
{ key: 'members', label: '成员与角色', adminOnly: true },
@@ -261,6 +263,7 @@ export function SystemSettingsView({ sources, connections, monitor, settings, pr
: <small>服务配置仅管理员可修改;当前值:测试地址 {settings.test_weknora_base_url || '未配置'} · 生产地址 {settings.production_weknora_base_url || '未配置'} · 生产 ID {settings.production_tenant_id || '未配置'} · 知识库 ID {settings.production_knowledge_base_id || '未配置'}。</small>}
</section> : null}
{section === 'bindings' ? <RoutingBindingsSection /> : null}
{section === 'production-bindings' ? <ProductionBindingsSection /> : null}
{section === 'storage' ? <section className="settings-panel wide storage-settings-panel" id="settings-storage">
<div className="settings-section-heading"><div><h2>存储与备份</h2><p>文件资源与生产发布前快照使用独立受限配置。</p></div></div>
<div className="storage-settings-grid"><dl><dt>Endpoint / Bucket</dt><dd>尚未配置</dd><dt>AccessKey / Secret</dt><dd>仅保存凭据引用,不可回读</dd><dt>CDN / 公开访问域名</dt><dd>尚未配置</dd><dt>临时签名有效期</dt><dd>待配置</dd></dl><dl><dt>测试备份前缀</dt><dd>/backup/weknora/test/</dd><dt>生产备份前缀</dt><dd>/backup/weknora/prod/</dd><dt>快照保留</dt><dd>最近 30 个生产快照</dd><dt>恢复验证</dt><dd>由生产门禁读回</dd></dl><ul className="target-list"><li>PostgreSQL <b>未配置</b></li><li>Elasticsearch <b>未配置</b></li><li>对象存储 <b>未配置</b></li><li>Neo4j <b>未配置</b></li><li>生产发布请求 <b>{productionEnabled ? '等待管理员核对' : '关闭'}</b></li><li>Release Worker <b>网页不可启用</b></li></ul></div>
+44
View File
@@ -0,0 +1,44 @@
import { describe, expect, it, vi } from 'vitest'
import { fetchProductionBindingOptions, fetchProductionBindings, saveProductionBindings } from './productionBindings'
const binding = {
internal_kb_id: '0b217e8a-2a65-40a3-ab8f-34ac5c7a6f05', external_kb_id: '',
internal_kb_name: '对内知识库-wiki', external_kb_name: '', updated_by: 'admin',
production_base_url: 'https://weknora-test.dobot.cn/', production_connected: true, production_message: '',
}
describe('production bindings API', () => {
it('reads the 源库 → 目标库 projection without leaking extra fields', async () => {
vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(JSON.stringify({ data: { ...binding, api_key: 'drop' } }), { status: 200 })))
const result = await fetchProductionBindings()
expect(result.internal_kb_name).toBe('对内知识库-wiki')
expect(result.production_connected).toBe(true)
expect(result).not.toHaveProperty('api_key')
vi.unstubAllGlobals()
})
it('lists production knowledge base options for the picker', async () => {
vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(JSON.stringify({ data: [{ id: 'kb-1', name: '对内知识库-wiki', knowledge_count: 2069 }, { id: '', name: '坏行' }] }), { status: 200 })))
expect(await fetchProductionBindingOptions()).toEqual([{ id: 'kb-1', name: '对内知识库-wiki', knowledge_count: 2069 }])
vi.unstubAllGlobals()
})
it('saves with an idempotency key and surfaces a rejected base', async () => {
const fetchMock = vi.fn()
.mockResolvedValueOnce(new Response(JSON.stringify({ data: binding }), { status: 200 }))
.mockResolvedValueOnce(new Response(JSON.stringify({ error: '对内绑定的生产知识库在该 WeKnora 实例中不存在' }), { status: 400 }))
vi.stubGlobal('fetch', fetchMock)
await expect(saveProductionBindings({ internal_kb_id: 'kb-1', external_kb_id: '' })).resolves.toMatchObject({ internal_kb_name: '对内知识库-wiki' })
const [, init] = fetchMock.mock.calls[0]
expect(init.method).toBe('PUT')
expect(init.headers['Idempotency-Key']).toBeTruthy()
await expect(saveProductionBindings({ internal_kb_id: 'kb-ghost', external_kb_id: '' })).rejects.toThrow('不存在')
vi.unstubAllGlobals()
})
it('reports an unverified save honestly instead of pretending it was checked', async () => {
vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(JSON.stringify({ data: { ...binding, production_connected: false, production_message: '生产 WeKnora 不可用:dial tcp: timeout' } }), { status: 200 })))
const result = await saveProductionBindings({ internal_kb_id: 'kb-1', external_kb_id: '' })
expect(result.production_connected).toBe(false)
expect(result.production_message).toContain('不可用')
vi.unstubAllGlobals()
})
})
+65
View File
@@ -0,0 +1,65 @@
import { newOperationID } from './operationKey'
import type { KnowledgeBaseOption } from './routingBindings'
// 源库 → 目标库(生产)映射:系统设置里选择发布要写入的生产知识库。
export interface ProductionBindings {
internal_kb_id: string
external_kb_id: string
internal_kb_name: string
external_kb_name: string
updated_by: string
production_base_url: string
production_connected: boolean
production_message: string
}
const record = (value: unknown): value is Record<string, unknown> => Boolean(value) && typeof value === 'object' && !Array.isArray(value)
function projection(item: Record<string, unknown>): ProductionBindings {
return {
internal_kb_id: typeof item.internal_kb_id === 'string' ? item.internal_kb_id : '',
external_kb_id: typeof item.external_kb_id === 'string' ? item.external_kb_id : '',
internal_kb_name: typeof item.internal_kb_name === 'string' ? item.internal_kb_name : '',
external_kb_name: typeof item.external_kb_name === 'string' ? item.external_kb_name : '',
updated_by: typeof item.updated_by === 'string' ? item.updated_by : '',
production_base_url: typeof item.production_base_url === 'string' ? item.production_base_url : '',
production_connected: item.production_connected === true,
production_message: typeof item.production_message === 'string' ? item.production_message : '',
}
}
export async function fetchProductionBindings(signal?: AbortSignal): Promise<ProductionBindings> {
const response = await fetch('/api/v1/production-kb-bindings', { signal, headers: { Accept: 'application/json' } })
if (!response.ok) throw new Error(`读取生产知识库绑定失败(HTTP ${response.status})`)
const value = await response.json() as unknown
if (!record(value) || !record(value.data)) throw new Error('生产知识库绑定响应无效')
return projection(value.data)
}
export async function fetchProductionBindingOptions(signal?: AbortSignal): Promise<KnowledgeBaseOption[]> {
const response = await fetch('/api/v1/production-kb-bindings/options', { signal, headers: { Accept: 'application/json' } })
if (!response.ok) throw new Error(`读取生产知识库列表失败(HTTP ${response.status})`)
const value = await response.json() as unknown
if (!record(value) || !Array.isArray(value.data)) throw new Error('生产知识库列表响应无效')
return value.data.filter(record).map((item) => ({
id: typeof item.id === 'string' ? item.id : '',
name: typeof item.name === 'string' ? item.name : '',
knowledge_count: typeof item.knowledge_count === 'number' ? item.knowledge_count : 0,
})).filter((item) => item.id)
}
export async function saveProductionBindings(input: { internal_kb_id: string; external_kb_id: string }, signal?: AbortSignal): Promise<ProductionBindings> {
const response = await fetch('/api/v1/production-kb-bindings', {
method: 'PUT', signal,
headers: { 'Content-Type': 'application/json', Accept: 'application/json', 'Idempotency-Key': newOperationID() },
body: JSON.stringify(input),
})
if (!response.ok) {
let reason = `保存生产知识库绑定失败(HTTP ${response.status})`
try { const body = await response.json() as unknown; if (record(body) && typeof body.error === 'string') reason = body.error } catch { /* keep status text */ }
throw new Error(reason)
}
const value = await response.json() as unknown
if (!record(value) || !record(value.data)) throw new Error('生产知识库绑定响应无效')
return projection(value.data)
}
+5
View File
@@ -11,6 +11,9 @@ export interface RoutingBindings {
external_kb_name: string
updated_by: string
weknora_connected: boolean
// 本次保存把多少条「还没上传」的待入库项重新指向了新库(读取时为 0)。
replanned: number
retired_duplicates: number
}
export interface KnowledgeBaseOption {
@@ -33,6 +36,8 @@ function bindingProjection(item: Record<string, unknown>): RoutingBindings {
external_kb_name: typeof item.external_kb_name === 'string' ? item.external_kb_name : '',
updated_by: typeof item.updated_by === 'string' ? item.updated_by : '',
weknora_connected: item.weknora_connected === true,
replanned: typeof item.replanned === 'number' ? item.replanned : 0,
retired_duplicates: typeof item.retired_duplicates === 'number' ? item.retired_duplicates : 0,
}
}