knowledge-governance/internal/httpapi/library_baseline.go

package httpapi

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"time"

	"knowledge-governance/internal/governance"
	"knowledge-governance/internal/jobs"
	postgresstore "knowledge-governance/internal/platform/postgres"
)

// 建立目标权威基线：首次发布时目标已经有内容，却没有生效版本记录。
//
// 唯一诚实的解法是"维护隔离 + 真实整库捕获 + 逐类校验 + 登记版本"：
//   - 不手写 production_effective，不伪造验收；
//   - 不清空目标换库绕过（那会把"非空但未知"变成"空"）；
//   - 页面只投递作业，停写入者、捕获、校验都在 worker 里执行。
//
// 这个入口只建作业，不签发发布授权：基线不等于发布，同步仍然要单独的 sync 授权。
const libraryBaselineJobType = "library.baseline"

type libraryBaselineRequest struct {
	Scope string `json:"scope"`
}

func libraryBaselineEnqueue(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
	}
	identity, ok := sessionFrom(r.Context())
	if !ok || roleRank(identity.Role) < roleRank(roleAdmin) {
		writeJSON(w, http.StatusForbidden, map[string]any{"error": "建立目标权威基线仅管理员可发起"})
		return
	}
	var request libraryBaselineRequest
	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
	}
	store, ok := libraryStore(config)
	if !ok {
		writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_baseline_unavailable",
			"error": "版本登记存储不可用"})
		return
	}
	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
	}
	for _, item := range versions {
		if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil {
			writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_exists",
				"error":       "该目标已有生效版本，不需要再建基线；换版本请走同步或恢复",
				"snapshot_id": item.SnapshotID})
			return
		}
	}
	// 目标必须真的读得到：读不到就不是"非空无权威版本"，而是另一个问题，
	// 不允许用建基线去掩盖连接/配置故障。
	probe := config.LibraryScopeProbe
	if probe == nil {
		probe = func(ctx context.Context, scope string) (bool, map[string]int64, error) {
			return productionScopeState(ctx, config, scope)
		}
	}
	empty, counts, probeErr := probe(r.Context(), request.Scope)
	if probeErr != nil {
		writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_baseline_unprobeable",
			"error": "无法读取目标整库状态，不能建立基线：" + probeErr.Error()})
		return
	}
	if empty {
		writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_target_empty",
			"error":  "目标整库为空，不需要基线：首次同步会在写入前生成完整预备份并登记生效版本",
			"counts": counts})
		return
	}
	// 维护互斥与同步一致：基线要在目标上开维护窗口，目标正在维护时直接拒绝。
	if body, held := instanceMaintenanceGuard(r.Context(), config, request.Scope); held {
		writeJSON(w, http.StatusConflict, body)
		return
	}
	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
		}
	}
	digest, err := governance.LibraryRequestDigest(request.Scope, governance.LibraryJobBaseline,
		map[string]any{"scope": request.Scope})
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest request"})
		return
	}
	now := time.Now().UTC()
	job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
		ID: fmt.Sprintf("libbase-%s-%d", request.Scope, now.UnixNano()), Scope: request.Scope,
		Kind:           governance.LibraryJobBaseline,
		IdempotencyKey: "library.baseline:" + request.Scope + ":" + now.Format("20060102T150405.000000000"),
		RequestDigest:  digest, Actor: identity.Username,
	}, governance.BaselineStages(), now)
	if errors.Is(err, governance.ErrLibraryJobActive) {
		writeJSON(w, http.StatusConflict, map[string]any{"code": "library_write_active",
			"error": "该目标已有在飞的写作业（同步/恢复/基线互斥），等它结束再发起"})
		return
	}
	if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
		writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_conflict",
			"error": "同一幂等键对应了不同的请求"})
		return
	}
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create library baseline job"})
		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})
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build baseline payload"})
		return
	}
	queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
		ID: governance.QueueJobIDFor(job.ID), Type: libraryBaselineJobType, Payload: payload,
		AvailableAt: now, MaxAttempts: 1, IdempotencyKey: "library.baseline.queue:" + job.ID,
	}, now)
	if err != nil || queueJob == nil {
		// 作业先落库、队列投递失败就把作业取消：不留下"看起来在跑"的空作业，
		// 否则它会一直占着该 scope 的写互斥。
		_ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败，未开始建立基线", now)
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue baseline 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 baseline job"})
		return
	}
	writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": false,
		"target_counts": counts})
}