Files
knowledge-governance/internal/httpapi/library_flow.go
T
姚章浩 dc9e827656
build / build (push) Failing after 17s
feat(library): 整库版本/作业持久化与状态读取契约,生产同步默认拒绝
- 新增 library_versions:测试封存、生产生效、生产预备份三类分开登记,
  生产生效版本唯一(旧版本必须先 supersede),测试快照不能顶替生产版本
- 新增 library_jobs + library_job_stages:业务作业身份、阶段、错误与恢复说明,
  分项计数用 NULL 表示"总量未知";互斥与幂等由数据库唯一索引保证
- 新增 GET /api/v1/library/overview、/jobs、/jobs/{id};POST /syncs 闸口未齐
  返回 503 缺项且不建作业
- 封存作业落库:先建业务作业再入队,worker 推进 export 阶段并登记 test_sealed
- 发布管理读真实状态:生产未知不再顶替、差异 unknown、启动同步按缺项禁用、
  同步明细/四类分项来自持久阶段;阶段勾选只认当前作业证据
- 准备信息标注封存时间与范围,不声称实时测试库已就绪
2026-10-28 13:58:51 +08:00

338 lines
14 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package httpapi
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"strings"
"time"
"knowledge-governance/internal/governance"
)
// 整库版本/状态/差异读取契约 + 生产同步的 fail-closed 入口。
//
// 三类版本严格分开,不能互相顶替:
// - 测试封存(archive 里的 sealed 产物)——只证明测试库有一份完整版本;
// - 生产当前生效(library_versions.kind=production_effective)——必须有生产验证证据;
// - 生产预备份(kind=production_prebackup)——每次生产写入前的独立备份。
// 生产未建立权威版本时返回 unknown,不拿最新测试快照冒充。
type libraryVersionView struct {
ID string `json:"id"`
Kind string `json:"kind"`
SnapshotID string `json:"snapshot_id"`
VersionLabel string `json:"version_label"`
SourceJobID string `json:"source_job_id"`
Evidence map[string]any `json:"evidence"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
SupersededAt *time.Time `json:"superseded_at,omitempty"`
}
type libraryKnownView struct {
State string `json:"state"` // known | unknown
Version *libraryVersionView `json:"version,omitempty"`
Reason string `json:"reason,omitempty"`
}
type libraryDiffView struct {
State string `json:"state"` // unknown | computed
Reason string `json:"reason,omitempty"`
Added []any `json:"added"`
Updated []any `json:"updated"`
Removed []any `json:"removed"`
}
type libraryGateView struct {
SyncAllowed bool `json:"sync_allowed"`
Reasons []string `json:"reasons"`
}
type libraryOverview struct {
Scope string `json:"scope"`
TestSealed *librarySnapshotView `json:"test_sealed,omitempty"`
TestSealedReason string `json:"test_sealed_reason,omitempty"`
Production libraryKnownView `json:"production"`
Prebackup libraryKnownView `json:"prebackup"`
Diff libraryDiffView `json:"diff"`
Job *governance.LibraryJob `json:"job,omitempty"`
Jobs []governance.LibraryJob `json:"jobs"`
Gate libraryGateView `json:"gate"`
CaptureAvailable bool `json:"capture_available"`
CaptureReason string `json:"capture_reason,omitempty"`
Now time.Time `json:"now"`
}
type libraryJobStore interface {
CreateLibraryJob(ctx context.Context, job governance.LibraryJob, stages []string, now time.Time) (governance.LibraryJob, bool, error)
LibraryJob(ctx context.Context, id string) (governance.LibraryJob, error)
ListLibraryJobs(ctx context.Context, scope string, limit int) ([]governance.LibraryJob, error)
RecordLibraryVersion(ctx context.Context, version governance.LibraryVersion) error
LibraryVersions(ctx context.Context, scope string, limit int) ([]governance.LibraryVersion, error)
UpdateLibraryStage(ctx context.Context, jobID, stage, state string, total, completed, bytesTotal, bytesDone *int64, detail string, now time.Time) error
AdvanceLibraryJob(ctx context.Context, jobID string, state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error
AttachLibraryQueueJob(ctx context.Context, jobID, queueJobID string, now time.Time) error
CancelLibraryJob(ctx context.Context, jobID, reason string, now time.Time) error
}
func libraryStore(config Config) (libraryJobStore, bool) {
store, ok := config.Store.(libraryJobStore)
return store, ok
}
func scopeFromQuery(r *http.Request) (string, error) {
scope := strings.TrimSpace(r.URL.Query().Get("scope"))
if scope == "" {
scope = "internal"
}
if scope != "internal" && scope != "external" {
return "", errors.New("scope must be internal or external")
}
return scope, nil
}
func libraryVersionViewOf(item governance.LibraryVersion) libraryVersionView {
return libraryVersionView{ID: item.ID, Kind: string(item.Kind), SnapshotID: item.SnapshotID,
VersionLabel: item.VersionLabel, SourceJobID: item.SourceJobID, Evidence: item.Evidence,
CreatedBy: item.CreatedBy, CreatedAt: item.CreatedAt, SupersededAt: item.SupersededAt}
}
// librarySyncGate 只依据真实存在的证据判断能不能写生产:闸口、生产端点、
// 生产权威版本、目标快照的预备份。任何一项缺失都给出具体原因,不默认放行。
func librarySyncGate(config Config, production libraryKnownView, prebackup libraryKnownView, target string) libraryGateView {
gate := libraryGateView{SyncAllowed: false, Reasons: []string{}}
if !config.ProductionReleaseRequestsEnabled {
gate.Reasons = append(gate.Reasons, "生产写入闸口关闭(PRODUCTION_RELEASE_REQUESTS_ENABLED=false)")
}
if !config.ProductionBackupRestoreVerified {
gate.Reasons = append(gate.Reasons, "没有已核验的生产恢复演练证据(PRODUCTION_BACKUP_RESTORE_VERIFIED=false)")
}
if strings.TrimSpace(config.ProductionWeKnoraBaseURL) == "" {
gate.Reasons = append(gate.Reasons, "生产端点未配置,无法核验生产当前版本")
}
if production.State != "known" {
gate.Reasons = append(gate.Reasons, "生产当前权威版本未知,不能启动写入")
}
if target == "" {
gate.Reasons = append(gate.Reasons, "还没有可同步的测试封存版本")
}
if prebackup.State != "known" {
gate.Reasons = append(gate.Reasons, "本次目标快照还没有生产预备份")
}
gate.SyncAllowed = len(gate.Reasons) == 0
return gate
}
func libraryOverviewHandler(w http.ResponseWriter, r *http.Request, config Config) {
scope, err := scopeFromQuery(r)
if err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
return
}
store, ok := libraryStore(config)
if !ok {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"available": false,
"reason": "版本登记存储不可用"})
return
}
now := time.Now().UTC()
overview := libraryOverview{Scope: scope, Jobs: []governance.LibraryJob{}, Now: now,
CaptureAvailable: strings.TrimSpace(config.LibraryAgentSocket) != "",
Gate: libraryGateView{Reasons: []string{}},
Diff: libraryDiffView{State: "unknown", Reason: "生产与测试的文档级差异读取未接线", Added: []any{}, Updated: []any{}, Removed: []any{}}}
if !overview.CaptureAvailable {
overview.CaptureReason = "主机侧封存 agent 未接线"
}
// 测试封存来自归档:它只是测试侧事实。
snapshots, snapErr := readLibrarySnapshots(config.LibraryArchiveRoot)
if snapErr != nil {
overview.TestSealedReason = "归档不可读"
} else if !snapshots.Available {
overview.TestSealedReason = snapshots.Reason
} else {
for index := range snapshots.Snapshots {
if snapshots.Snapshots[index].Scope == scope {
overview.TestSealed = &snapshots.Snapshots[index]
break
}
}
if overview.TestSealed == nil {
overview.TestSealedReason = "还没有该目标的测试封存"
}
}
// 生产生效 / 预备份 / 作业来自持久登记。
versions, err := store.LibraryVersions(r.Context(), scope, 50)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library versions"})
return
}
overview.Production = libraryKnownView{State: "unknown", Reason: "还没有生产验证通过的生效版本记录"}
overview.Prebackup = libraryKnownView{State: "unknown", Reason: "本次目标快照还没有生产预备份"}
target := ""
if overview.TestSealed != nil {
target = overview.TestSealed.SnapshotID
}
for index := range versions {
item := versions[index]
view := libraryVersionViewOf(item)
switch item.Kind {
case governance.LibraryVersionProductionEffective:
if item.SupersededAt == nil {
overview.Production = libraryKnownView{State: "known", Version: &view}
}
case governance.LibraryVersionProductionPrebackup:
// 只有与目标快照绑定的预备份才算"这次同步前的备份"。
if target != "" && item.SnapshotID == target {
overview.Prebackup = libraryKnownView{State: "known", Version: &view}
}
}
}
recent, err := store.ListLibraryJobs(r.Context(), scope, 20)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library jobs"})
return
}
overview.Jobs = recent
for index := range recent {
if recent[index].State == governance.LibraryJobPending || recent[index].State == governance.LibraryJobRunning {
overview.Job = &recent[index]
break
}
}
if overview.Job == nil && len(recent) > 0 {
overview.Job = &recent[0]
}
overview.Gate = librarySyncGate(config, overview.Production, overview.Prebackup, target)
writeJSON(w, http.StatusOK, overview)
}
func libraryJobsHandler(w http.ResponseWriter, r *http.Request, config Config) {
scope, err := scopeFromQuery(r)
if err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
return
}
store, ok := libraryStore(config)
if !ok {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"available": false, "reason": "版本登记存储不可用", "jobs": []any{}})
return
}
items, err := store.ListLibraryJobs(r.Context(), scope, 20)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library jobs"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"available": true, "jobs": items})
}
func libraryJobDetailHandler(w http.ResponseWriter, r *http.Request, config Config) {
store, ok := libraryStore(config)
if !ok {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"available": false, "reason": "版本登记存储不可用"})
return
}
id := strings.TrimSpace(r.PathValue("id"))
if id == "" {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "job id required"})
return
}
job, err := store.LibraryJob(r.Context(), id)
if err != nil {
writeJSON(w, http.StatusNotFound, map[string]any{"available": false, "reason": "作业不存在"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"available": true, "job": job})
}
// librarySyncEnqueue 是生产同步的唯一入口:闸口未齐时**不建作业**,直接说明缺项。
// 这样按钮永远不会假接通,也不会留下一个永远失败的空作业。
func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
if r.Method != http.MethodPost {
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
return
}
store, ok := libraryStore(config)
if !ok {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_sync_unavailable", "reasons": []string{"版本登记存储不可用"}})
return
}
var request struct {
Scope string `json:"scope"`
Target string `json:"target_snapshot_id"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&request); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
return
}
if request.Scope != "internal" && request.Scope != "external" {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "scope must be internal or external"})
return
}
overview := libraryOverview{Scope: request.Scope}
versions, err := store.LibraryVersions(r.Context(), request.Scope, 50)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library versions"})
return
}
overview.Production = libraryKnownView{State: "unknown"}
overview.Prebackup = libraryKnownView{State: "unknown"}
for index := range versions {
item := versions[index]
if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil {
view := libraryVersionViewOf(item)
overview.Production = libraryKnownView{State: "known", Version: &view}
}
if item.Kind == governance.LibraryVersionProductionPrebackup && request.Target != "" && item.SnapshotID == request.Target {
view := libraryVersionViewOf(item)
overview.Prebackup = libraryKnownView{State: "known", Version: &view}
}
}
gate := librarySyncGate(config, overview.Production, overview.Prebackup, request.Target)
if !gate.SyncAllowed {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{
"code": "library_sync_blocked", "reasons": gate.Reasons,
"production": overview.Production, "prebackup": overview.Prebackup,
})
return
}
// 走到这里说明闸口与证据都齐了才真正建作业;互斥与幂等由数据库索引保证。
digest, err := governance.LibraryRequestDigest(request.Scope, governance.LibraryJobSync,
map[string]any{"target_snapshot_id": request.Target})
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest request"})
return
}
actor := ""
if identity, ok := sessionFrom(r.Context()); ok {
actor = identity.Username
}
now := time.Now().UTC()
jobID := fmt.Sprintf("libsync-%s-%d", request.Scope, now.UnixNano())
job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
ID: jobID, Scope: request.Scope, Kind: governance.LibraryJobSync,
IdempotencyKey: "library.sync:" + request.Scope + ":" + request.Target, RequestDigest: digest,
TargetSnapshotID: request.Target, Actor: actor,
}, governance.SyncStages(), now)
if errors.Is(err, governance.ErrLibraryJobActive) {
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_active", "error": "该目标已有在飞的同步作业"})
return
}
if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_conflict", "error": "同一幂等键对应了不同的请求"})
return
}
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create library sync job"})
return
}
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": reused})
}