姚章浩
288e01a01a
build / build (push) Failing after 16s
为什么：用户确认维护窗口方案，需要真实占有目标而不是 503 页面。 - 迁移 000070 library_maintenance：scope 唯一持有者、作业/目标身份、服务原状态、阶段。 - worker 在写目标前 acquire，验证+登记成功才 release；回滚失败/未验证保持 needs_restore。 - API GET /library/maintenance + 同步入口在他人持有时 409。 - 已知缺陷如实记录：候选实例该路由 404；kill 取证脚本竞态导致三项断言为 false。
689 lines
28 KiB
Go
689 lines
28 KiB
Go
package httpapi
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
"knowledge-governance/internal/jobs"
|
||
"knowledge-governance/internal/librarydiff"
|
||
"knowledge-governance/internal/librarysnapshot"
|
||
postgresstore "knowledge-governance/internal/platform/postgres"
|
||
)
|
||
|
||
// 整库版本/状态/差异读取契约 + 生产同步的 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"`
|
||
}
|
||
|
||
// librarySyncJobType 与 worker 的处理器类型一致；单独定义避免 API 依赖 worker 包。
|
||
const librarySyncJobType = "library.sync"
|
||
|
||
type libraryKnownView struct {
|
||
State string `json:"state"` // known | unknown
|
||
Version *libraryVersionView `json:"version,omitempty"`
|
||
Reason string `json:"reason,omitempty"`
|
||
// EmptyBaseline 表示"没有生效版本记录，但生产目标可读且确实是空的"：
|
||
// 这是首次发布的合法起点，和"生产非空却没有权威版本"必须区分。
|
||
EmptyBaseline bool `json:"empty_baseline,omitempty"`
|
||
}
|
||
|
||
// productionScopeEmpty 只读探测生产目标：真正连库数一遍，不用缓存或推断。
|
||
func productionScopeEmpty(ctx context.Context, config Config, scope string) (bool, error) {
|
||
if strings.TrimSpace(config.LibraryProductionDatabaseURL) == "" {
|
||
return false, errors.New("生产整库目标未配置")
|
||
}
|
||
kb := config.LibraryProductionInternalKB
|
||
if scope == "external" {
|
||
kb = config.LibraryProductionExternalKB
|
||
}
|
||
if kb == "" {
|
||
return false, errors.New("生产知识库未配置")
|
||
}
|
||
pool, err := pgxpool.New(ctx, config.LibraryProductionDatabaseURL)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
defer pool.Close()
|
||
var count int64
|
||
err = pool.QueryRow(ctx, `SELECT count(*) FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND deleted_at IS NULL`,
|
||
config.LibraryProductionTenantID, kb).Scan(&count)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return count == 0, nil
|
||
}
|
||
|
||
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"`
|
||
// BaseSnapshot/TargetSnapshot 标明这次比对用的是哪两个整库版本，
|
||
// 避免把某次历史差异当成"当前拟发布"。
|
||
BaseSnapshot string `json:"base_snapshot,omitempty"`
|
||
TargetSnapshot string `json:"target_snapshot,omitempty"`
|
||
// Wiki 是**页面级**差异：只能证明页面被增/改/撤，不能证明共享正文、实体、
|
||
// 概念或边被重新计算——后者需要 WeKnora 内部语义，本平台不做也不声称。
|
||
Wiki *libraryWikiDiffView `json:"wiki,omitempty"`
|
||
}
|
||
|
||
type libraryWikiDiffView struct {
|
||
State string `json:"state"` // computed | unknown
|
||
Reason string `json:"reason,omitempty"`
|
||
Added int `json:"added"`
|
||
Updated int `json:"updated"`
|
||
Removed int `json:"removed"`
|
||
// Sample 只列前若干页，避免把整库 Wiki 列表塞进接口。
|
||
Sample []string `json:"sample"`
|
||
Note string `json:"note"`
|
||
}
|
||
|
||
// wikiDiffOf 比较两份快照的 wiki_pages 行数组（页面级）。
|
||
// 它不能替代"删除文档后共享正文/实体/边/引用如何变化"的语义证据。
|
||
func wikiDiffOf(archive *librarysnapshot.Archive, baseID, targetID string) *libraryWikiDiffView {
|
||
unknown := func(reason string) *libraryWikiDiffView {
|
||
return &libraryWikiDiffView{State: "unknown", Reason: reason, Sample: []string{},
|
||
Note: "页面级比对缺失时不能声称 Wiki 已同步"}
|
||
}
|
||
rows := func(id string) ([]map[string]any, string) {
|
||
manifest, err := archive.Load(id)
|
||
if err != nil {
|
||
return nil, "快照不可读"
|
||
}
|
||
for _, part := range manifest.Parts {
|
||
if part.Kind != "wiki" || part.Name != "wiki_pages" {
|
||
continue
|
||
}
|
||
reader, err := archive.OpenPart(part)
|
||
if err != nil {
|
||
return nil, "Wiki 正文不可读"
|
||
}
|
||
defer reader.Close()
|
||
parsed, err := librarydiff.ReadDocuments(reader)
|
||
if err != nil {
|
||
return nil, "Wiki 行格式不符"
|
||
}
|
||
return parsed, ""
|
||
}
|
||
return nil, "快照里没有 wiki_pages 部件"
|
||
}
|
||
base, reason := rows(baseID)
|
||
if reason != "" {
|
||
return unknown("生产版本：" + reason)
|
||
}
|
||
target, reason := rows(targetID)
|
||
if reason != "" {
|
||
return unknown("测试版本：" + reason)
|
||
}
|
||
result, err := librarydiff.Compare(base, target)
|
||
if err != nil {
|
||
return unknown("Wiki 页面比对失败：" + err.Error())
|
||
}
|
||
view := &libraryWikiDiffView{State: "computed", Added: len(result.Added), Updated: len(result.Updated),
|
||
Removed: len(result.Removed), Sample: []string{},
|
||
Note: "页面级差异：只说明页面被增/改/撤，不代表实体、概念、边或引用已重算"}
|
||
for _, item := range result.Added {
|
||
if len(view.Sample) >= 20 {
|
||
break
|
||
}
|
||
view.Sample = append(view.Sample, "新增 "+item.Title)
|
||
}
|
||
for _, item := range result.Updated {
|
||
if len(view.Sample) >= 20 {
|
||
break
|
||
}
|
||
view.Sample = append(view.Sample, "更新 "+item.Title)
|
||
}
|
||
for _, item := range result.Removed {
|
||
if len(view.Sample) >= 20 {
|
||
break
|
||
}
|
||
view.Sample = append(view.Sample, "撤销 "+item.Title)
|
||
}
|
||
return view
|
||
}
|
||
|
||
// computeLibraryDiff 用归档里的 knowledges 行数组比较两个快照：
|
||
// 身份是文档 ID，内容是排除易变列后的整行摘要；任一侧读不到就返回 false，
|
||
// 由调用方保持 unknown 并写明原因，而不是给出空差异冒充"无变化"。
|
||
func computeLibraryDiff(root, baseID, targetID string) (libraryDiffView, bool) {
|
||
unknown := libraryDiffView{State: "unknown", Added: []any{}, Updated: []any{}, Removed: []any{},
|
||
BaseSnapshot: baseID, TargetSnapshot: targetID}
|
||
if root == "" {
|
||
unknown.Reason = "整库归档未配置，无法比较生产与测试版本"
|
||
return unknown, false
|
||
}
|
||
if baseID == "" || targetID == "" {
|
||
unknown.Reason = "生产当前版本或测试封存缺失，没有可比的两份整库版本"
|
||
return unknown, false
|
||
}
|
||
archive, err := librarysnapshot.Open(root, 256*1024*1024)
|
||
if err != nil {
|
||
unknown.Reason = "归档根不可用，无法比较"
|
||
return unknown, false
|
||
}
|
||
defer archive.Close()
|
||
read := func(id string) ([]map[string]any, string) {
|
||
manifest, loadErr := archive.Load(id)
|
||
if loadErr != nil {
|
||
return nil, "快照不在本地归档或校验失败"
|
||
}
|
||
for _, part := range manifest.Parts {
|
||
if part.Kind != "database" || part.Name != "knowledges" {
|
||
continue
|
||
}
|
||
reader, openErr := archive.OpenPart(part)
|
||
if openErr != nil {
|
||
return nil, "归档正文不可读"
|
||
}
|
||
defer reader.Close()
|
||
rows, readErr := librarydiff.ReadDocuments(reader)
|
||
if readErr != nil {
|
||
return nil, "归档正文格式不符"
|
||
}
|
||
return rows, ""
|
||
}
|
||
return nil, "快照里没有 knowledges 部件"
|
||
}
|
||
base, reason := read(baseID)
|
||
if reason != "" {
|
||
unknown.Reason = "生产版本：" + reason
|
||
return unknown, false
|
||
}
|
||
target, reason := read(targetID)
|
||
if reason != "" {
|
||
unknown.Reason = "测试版本：" + reason
|
||
return unknown, false
|
||
}
|
||
result, err := librarydiff.Compare(base, target)
|
||
if err != nil {
|
||
unknown.Reason = "差异计算失败：" + err.Error()
|
||
return unknown, false
|
||
}
|
||
view := libraryDiffView{State: "computed", Added: []any{}, Updated: []any{}, Removed: []any{},
|
||
BaseSnapshot: baseID, TargetSnapshot: targetID}
|
||
view.Wiki = wikiDiffOf(archive, baseID, targetID)
|
||
for _, item := range result.Added {
|
||
view.Added = append(view.Added, item)
|
||
}
|
||
for _, item := range result.Updated {
|
||
view.Updated = append(view.Updated, item)
|
||
}
|
||
for _, item := range result.Removed {
|
||
view.Removed = append(view.Removed, item)
|
||
}
|
||
return view, true
|
||
}
|
||
|
||
// libraryAuthorizationView 让页面能直说"有没有授权、有效期到什么时候"。
|
||
type libraryAuthorizationView struct {
|
||
State string `json:"state"` // active | none
|
||
AuthorizedBy string `json:"authorized_by,omitempty"`
|
||
ExpiresAt *time.Time `json:"expires_at,omitempty"`
|
||
Reason string `json:"reason,omitempty"`
|
||
}
|
||
|
||
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"`
|
||
Authorization libraryAuthorizationView `json:"authorization"`
|
||
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
|
||
}
|
||
|
||
// activeGrantFor 只查"目标+快照+同一参数摘要"的未过期未撤销授权。
|
||
func activeGrantFor(ctx context.Context, config Config, scope, snapshotID string) (governance.LibraryReleaseAuthorization, bool) {
|
||
if snapshotID == "" {
|
||
return governance.LibraryReleaseAuthorization{}, false
|
||
}
|
||
store, ok := config.Store.(interface {
|
||
ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, paramsDigest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
|
||
})
|
||
if !ok {
|
||
return governance.LibraryReleaseAuthorization{}, false
|
||
}
|
||
digest, err := librarySyncParamsDigest(scope, snapshotID)
|
||
if err != nil {
|
||
return governance.LibraryReleaseAuthorization{}, false
|
||
}
|
||
grant, found, err := store.ActiveLibraryAuthorization(ctx, scope, snapshotID, digest, time.Now().UTC())
|
||
if err != nil || !found {
|
||
return governance.LibraryReleaseAuthorization{}, false
|
||
}
|
||
return grant, true
|
||
}
|
||
|
||
func authorizationViewOf(grant governance.LibraryReleaseAuthorization, found bool) libraryAuthorizationView {
|
||
if !found {
|
||
return libraryAuthorizationView{State: "none", Reason: "没有与目标快照匹配的有效发布授权"}
|
||
}
|
||
expires := grant.ExpiresAt
|
||
return libraryAuthorizationView{State: "active", AuthorizedBy: grant.AuthorizedBy, ExpiresAt: &expires}
|
||
}
|
||
|
||
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 errString(err error) string {
|
||
if err == nil {
|
||
return ""
|
||
}
|
||
return err.Error()
|
||
}
|
||
|
||
func librarySyncGate(config Config, production libraryKnownView, prebackup libraryKnownView, target string) libraryGateView {
|
||
return librarySyncGateWithGrant(config, production, prebackup, target, false)
|
||
}
|
||
|
||
// librarySyncGateWithGrant 在原有证据之外再要求一份有效授权：
|
||
// 授权绑定了目标、快照、参数与有效期，缺它就不能写生产。
|
||
func librarySyncGateWithGrant(config Config, production libraryKnownView, prebackup libraryKnownView, target string, hasGrant bool) libraryGateView {
|
||
gate := libraryGateView{SyncAllowed: false, Reasons: []string{}}
|
||
if !hasGrant {
|
||
gate.Reasons = append(gate.Reasons, "没有与目标快照匹配的有效发布授权")
|
||
}
|
||
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）")
|
||
}
|
||
// 生产端点以"整库目标是否配置"为准：整库同步直接写 PG/ES/对象，
|
||
// 不依赖 WeKnora 的 HTTP 端，所以不能拿它是否配置当门禁。
|
||
if strings.TrimSpace(config.LibraryProductionDatabaseURL) == "" {
|
||
gate.Reasons = append(gate.Reasons, "生产整库目标未配置（LIBRARY_PRODUCTION_DATABASE_URL），无法核验生产当前状态")
|
||
}
|
||
// 首次发布：生产还没有"生效版本"记录，但只要生产目标可读且是空的，
|
||
// 这就是一个合法的"空基线"，不应该被永久挡住。非空但无权威版本仍然阻断。
|
||
if production.State != "known" && !production.EmptyBaseline {
|
||
gate.Reasons = append(gate.Reasons, "生产当前权威版本未知，不能启动写入")
|
||
}
|
||
if target == "" {
|
||
gate.Reasons = append(gate.Reasons, "还没有可同步的测试封存版本")
|
||
}
|
||
// 预备份是作业内第一步生成的：启动前要求它已存在会造成死循环。
|
||
// 这里只在作业已经被判定失败或回滚失败时提示，不作为启动前的硬门禁。
|
||
if prebackup.State == "failed" {
|
||
gate.Reasons = append(gate.Reasons, "上一次预备份失败，先处理后才能重试")
|
||
}
|
||
if config.LibraryCapacityBudgetBytes <= 0 {
|
||
gate.Reasons = append(gate.Reasons, "容量预算未设置（LIBRARY_CAPACITY_BUDGET_BYTES），无法证明装得下")
|
||
}
|
||
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]
|
||
}
|
||
|
||
// 差异是真实计算：生产权威版本 + 测试封存都在本地归档时给出逐篇增改撤销，
|
||
// 否则保持 unknown 并说明缺哪一侧，不用空数组冒充"无变化"。
|
||
baseSnapshot := ""
|
||
if overview.Production.State == "known" && overview.Production.Version != nil {
|
||
baseSnapshot = overview.Production.Version.SnapshotID
|
||
}
|
||
// 归档可能很大：差异计算放进看门狗，超时就保持 unknown，别让页面请求挂死。
|
||
diffResult := make(chan struct {
|
||
view libraryDiffView
|
||
ok bool
|
||
}, 1)
|
||
go func() {
|
||
view, ok := computeLibraryDiff(config.LibraryArchiveRoot, baseSnapshot, target)
|
||
diffResult <- struct {
|
||
view libraryDiffView
|
||
ok bool
|
||
}{view, ok}
|
||
}()
|
||
var view libraryDiffView
|
||
var diffOK bool
|
||
select {
|
||
case outcome := <-diffResult:
|
||
view, diffOK = outcome.view, outcome.ok
|
||
case <-time.After(8 * time.Second):
|
||
view = libraryDiffView{State: "unknown", Reason: "差异计算超时（归档较大），本次保持不变更结论",
|
||
Added: []any{}, Updated: []any{}, Removed: []any{}, BaseSnapshot: baseSnapshot, TargetSnapshot: target}
|
||
diffOK = false
|
||
}
|
||
if diffOK || view.Reason != "" {
|
||
overview.Diff = view
|
||
}
|
||
|
||
grant, hasGrant := activeGrantFor(r.Context(), config, scope, target)
|
||
overview.Authorization = authorizationViewOf(grant, hasGrant)
|
||
overview.Gate = librarySyncGateWithGrant(config, overview.Production, overview.Prebackup, target, hasGrant)
|
||
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"}
|
||
if empty, probeErr := productionScopeEmpty(r.Context(), config, request.Scope); probeErr == nil && empty {
|
||
overview.Production = libraryKnownView{State: "unknown", EmptyBaseline: true,
|
||
Reason: "生产目标可读且为空：本次是首次发布，写入前仍会生成完整预备份"}
|
||
}
|
||
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}
|
||
}
|
||
}
|
||
grantFound, hasGrant := activeGrantFor(r.Context(), config, request.Scope, request.Target)
|
||
_ = grantFound
|
||
// 目标正在维护（被别的作业持有）：直接拒绝新写入，不等到 worker 里才发现。
|
||
if maintenanceStore, ok := config.Store.(interface {
|
||
LibraryMaintenanceOf(ctx context.Context, scope string) (*postgresstore.LibraryMaintenance, error)
|
||
}); ok {
|
||
if held, err := maintenanceStore.LibraryMaintenanceOf(r.Context(), request.Scope); err == nil && held != nil && held.ReleasedAt == nil {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_target_in_maintenance",
|
||
"error": "目标正在维护中，验证完成前不接受新的写入请求",
|
||
"owner_job": held.OwnerJobID, "state": held.State})
|
||
return
|
||
}
|
||
}
|
||
gate := librarySyncGateWithGrant(config, overview.Production, overview.Prebackup, request.Target, hasGrant)
|
||
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
|
||
}
|
||
// 上一次同参数请求失败/被取消后，幂等键不能再把新请求塞回那个死作业里：
|
||
// 换一个"重试代次"键重新建作业，旧作业保留为审计记录。
|
||
if reused && (job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled) {
|
||
job, reused, err = store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||
ID: fmt.Sprintf("libsync-%s-%d", request.Scope, now.UnixNano()), Scope: request.Scope,
|
||
Kind: governance.LibraryJobSync,
|
||
IdempotencyKey: "library.sync:" + request.Scope + ":" + request.Target + ":retry:" + now.Format("20060102T150405.000000000"),
|
||
RequestDigest: digest, TargetSnapshotID: request.Target, Actor: actor,
|
||
}, governance.SyncStages(), now)
|
||
if err != nil || reused {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_retry_conflict",
|
||
"error": "无法为重试创建新作业", "detail": errString(err)})
|
||
return
|
||
}
|
||
}
|
||
if reused {
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": true})
|
||
return
|
||
}
|
||
// 业务作业先落库，再投递队列：投递失败就把作业标失败，不留下"看起来在跑"的空作业。
|
||
payload, err := json.Marshal(map[string]string{
|
||
"library_job_id": job.ID, "scope": request.Scope, "target_snapshot_id": request.Target,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build sync payload"})
|
||
return
|
||
}
|
||
queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
|
||
ID: governance.QueueJobIDFor(job.ID), Type: librarySyncJobType, Payload: payload, AvailableAt: now,
|
||
MaxAttempts: 1, IdempotencyKey: "library.sync.queue:" + job.ID,
|
||
}, now)
|
||
if err != nil || queueJob == nil {
|
||
log.Printf("library sync enqueue failed job=%s: %v", job.ID, err)
|
||
_ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败，未开始同步", now)
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue sync job"})
|
||
return
|
||
}
|
||
if attachErr := store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now); attachErr != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach sync job"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": false})
|
||
}
|