姚章浩
2619 lines
116 KiB
Go
2619 lines
116 KiB
Go
package httpapi
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"crypto/subtle"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"knowledge-governance/internal/backuphost"
|
||
"knowledge-governance/internal/governance"
|
||
"knowledge-governance/internal/jobs"
|
||
"knowledge-governance/internal/monitoring"
|
||
postgresstore "knowledge-governance/internal/platform/postgres"
|
||
)
|
||
|
||
type JobStore interface {
|
||
Enqueue(ctx context.Context, request jobs.EnqueueRequest, now time.Time) (*jobs.Job, error)
|
||
List(ctx context.Context, limit int) ([]jobs.Job, error)
|
||
}
|
||
|
||
type MonitorProvider interface {
|
||
Overview(ctx context.Context) (governance.MonitorOverview, error)
|
||
ListFiles(ctx context.Context, knowledgeBaseID, status string, limit, offset int) (governance.KnowledgeMonitorFilePage, error)
|
||
Metrics(ctx context.Context) (governance.RunMetrics, error)
|
||
Services(ctx context.Context) []governance.ServiceHealth
|
||
}
|
||
|
||
type Store interface {
|
||
JobStore
|
||
ListSourceConnections(ctx context.Context) (governance.SourceConnectionPage, error)
|
||
CreateSourceSettingsRevisionWithOperation(ctx context.Context, connectionID string, draft postgresstore.SourceSettingsDraft, operationID string, now time.Time) (string, bool, error)
|
||
ActivateSourceSettingsRevisionWithOperation(ctx context.Context, connectionID, revisionID, operationID string, now time.Time) (bool, error)
|
||
ListSourceConnectionRevisions(ctx context.Context, connectionID string) ([]governance.SourceConnectionRevision, error)
|
||
TriggerPPSourceScan(ctx context.Context, connectionID string, now time.Time) (governance.SourceScan, error)
|
||
TriggerPPSourceScanWithOperation(ctx context.Context, connectionID string, now time.Time, operationID string) (governance.SourceScan, bool, error)
|
||
ListSourceScans(ctx context.Context, connectionID string, limit, offset int) (governance.SourceScanPage, error)
|
||
GetSourceScanDetail(ctx context.Context, connectionID, scanID string) (governance.SourceScanDetailView, error)
|
||
ListReviewCandidates(ctx context.Context, filter governance.ReviewCandidateFilter) (governance.ReviewCandidatePage, error)
|
||
ReviewPublication(ctx context.Context, request governance.ReviewPublicationRequest) (governance.PublicationReview, error)
|
||
ReviewPublicationWithOperation(ctx context.Context, request governance.ReviewPublicationRequest, operationID string) (governance.PublicationReview, bool, error)
|
||
CreateReleaseBatch(ctx context.Context, request governance.CreateReleaseBatchRequest) (governance.ReleaseBatch, error)
|
||
CreateReleaseBatchWithOperation(ctx context.Context, request governance.CreateReleaseBatchRequest, operationID string) (governance.ReleaseBatch, bool, error)
|
||
GetReleaseAuditChain(ctx context.Context, executionID string) (governance.ReleaseAuditChain, error)
|
||
GenerateReleaseChangeSetWithOperation(ctx context.Context, request governance.GenerateReleaseChangeSetRequest, operationID string) (governance.ReleaseChangeSet, bool, error)
|
||
ConfirmReleaseChangeSetWithOperation(ctx context.Context, request governance.ConfirmReleaseChangeSetRequest, operationID string) (governance.ReleaseChangeSet, bool, error)
|
||
ListReleaseChangeSets(ctx context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseChangeSetPage, error)
|
||
GetReleaseChangeSet(ctx context.Context, id string) (governance.ReleaseChangeSetDetail, error)
|
||
LatestReleaseChangeSet(ctx context.Context) (governance.ReleaseChangeSetDetail, error)
|
||
SealReleaseBatch(ctx context.Context, request governance.SealReleaseBatchRequest) (governance.ReleaseBatch, error)
|
||
SealReleaseBatchWithOperation(ctx context.Context, request governance.SealReleaseBatchRequest, operationID string) (governance.ReleaseBatch, bool, error)
|
||
ListReleaseBatches(ctx context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseBatchPage, error)
|
||
GetReleaseBatch(ctx context.Context, batchID string) (governance.ReleaseBatchDetail, error)
|
||
AuthorizeReleaseBatch(ctx context.Context, request governance.AuthorizeReleaseBatchRequest) (governance.ReleaseExecution, error)
|
||
AuthorizeReleaseBatchWithOperation(ctx context.Context, request governance.AuthorizeReleaseBatchRequest, operationID string) (governance.ReleaseExecution, bool, error)
|
||
AuthorizeReleaseRollback(ctx context.Context, request governance.AuthorizeReleaseRollbackRequest) (governance.ReleaseExecution, error)
|
||
AuthorizeReleaseRollbackWithOperation(ctx context.Context, request governance.AuthorizeReleaseRollbackRequest, operationID string) (governance.ReleaseExecution, bool, error)
|
||
RetryReleaseExecution(ctx context.Context, request governance.RetryReleaseExecutionRequest) (governance.ReleaseExecution, error)
|
||
RetryReleaseExecutionWithOperation(ctx context.Context, request governance.RetryReleaseExecutionRequest, operationID string) (governance.ReleaseExecution, bool, error)
|
||
ListReleaseExecutions(ctx context.Context, filter governance.ReleaseExecutionFilter) (governance.ReleaseExecutionPage, error)
|
||
GetReleaseExecution(ctx context.Context, executionID string) (governance.ReleaseExecutionDetail, error)
|
||
ListReleasePlan(ctx context.Context) (governance.ReleasePlan, error)
|
||
GetActiveUISettings(ctx context.Context) (governance.UISettings, error)
|
||
GetUISettingsRevision(ctx context.Context, revisionID string) (governance.UISettings, error)
|
||
CreateUISettingsCandidate(ctx context.Context, request governance.CreateUISettingsCandidateRequest) (governance.UISettings, error)
|
||
CreateUISettingsCandidateWithOperation(ctx context.Context, request governance.CreateUISettingsCandidateRequest, operationID string) (governance.UISettings, bool, error)
|
||
ActivateUISettings(ctx context.Context, revisionID string, now time.Time) (governance.UISettings, error)
|
||
ActivateUISettingsWithOperation(ctx context.Context, revisionID string, now time.Time, operationID string) (governance.UISettings, bool, error)
|
||
ListSourceAssets(ctx context.Context, filter governance.SourceAssetFilter) (governance.SourceAssetPage, error)
|
||
ListCatalogSources(ctx context.Context) ([]governance.CatalogSource, error)
|
||
ListCatalogNodes(ctx context.Context, filter governance.CatalogNodeFilter) (governance.CatalogNodePage, error)
|
||
ListCatalogSnapshots(ctx context.Context, connectionID string, limit, offset int) (governance.CatalogSnapshotPage, error)
|
||
ListCatalogNodesInSnapshot(ctx context.Context, filter governance.CatalogSnapshotNodeFilter) (governance.CatalogNodePage, error)
|
||
GetLatestCatalogChangeSet(ctx context.Context, filter governance.CatalogChangeSetFilter) (governance.CatalogChangeSetDetail, error)
|
||
RouteCatalogFiles(ctx context.Context, request governance.CatalogRouteRequest) (governance.CatalogRouteResult, error)
|
||
RouteCatalogFilesWithOperation(ctx context.Context, request governance.CatalogRouteRequest, operationID string) (governance.CatalogRouteResult, bool, error)
|
||
EnqueueCatalogRefreshWithOperation(ctx context.Context, source governance.CatalogSource, operationID string, now time.Time) (jobs.Job, bool, error)
|
||
ListScopeRules(ctx context.Context, connectionID string) ([]governance.ScopeRule, error)
|
||
MutateScopeRule(ctx context.Context, request governance.ScopeRuleMutation) (governance.ScopeRule, error)
|
||
MutateScopeRuleWithOperation(ctx context.Context, request governance.ScopeRuleMutation, operationID string) (governance.ScopeRule, bool, error)
|
||
ListFilenameFilterRules(ctx context.Context, connectionID string) ([]governance.FilenameFilterRule, error)
|
||
MutateFilenameFilterRuleWithOperation(ctx context.Context, request governance.FilenameFilterMutation, operationID string) (governance.FilenameFilterRule, bool, error)
|
||
ListTestPublications(ctx context.Context, filter governance.TestPublicationFilter) (governance.TestPublicationPage, error)
|
||
StartTestPublications(ctx context.Context, request governance.StartTestPublicationsRequest) (governance.StartTestPublicationsResult, error)
|
||
StartTestPublicationsWithOperation(ctx context.Context, request governance.StartTestPublicationsRequest, operationID string) (governance.StartTestPublicationsResult, bool, error)
|
||
ConfirmCatalogChangeSetWithOperation(ctx context.Context, request governance.ConfirmCatalogChangeSetRequest, operationID string) (governance.ConfirmCatalogChangeSetResult, bool, error)
|
||
ListAcceptanceResults(ctx context.Context, publicationID string) ([]governance.AcceptanceRecord, error)
|
||
RetryPublication(ctx context.Context, publicationID string, now time.Time) error
|
||
RetryPublicationWithOperation(ctx context.Context, publicationID string, now time.Time, operationID string) (bool, error)
|
||
ListJobFailures(ctx context.Context, filter governance.JobFailureFilter) (governance.JobFailurePage, error)
|
||
RetryPublicationsBatchWithOperation(ctx context.Context, request governance.RetryPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error)
|
||
CancelPublicationsBatchWithOperation(ctx context.Context, request governance.CancelPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error)
|
||
RequeueArtifactFetchWithOperation(ctx context.Context, request governance.RequeueArtifactFetchRequest, operationID string) (governance.RequeueArtifactFetchResult, bool, error)
|
||
}
|
||
|
||
type Config struct {
|
||
QAOrg QAOrgStore
|
||
QAOps QAOpsReader
|
||
QAOpsTenantKey string
|
||
QAOpsEnabled bool
|
||
QAOpsProblem string
|
||
Store Store
|
||
Monitor MonitorProvider
|
||
InternalRoutingKnowledgeBaseID string
|
||
ExternalRoutingKnowledgeBaseID string
|
||
RoutingBindings RoutingBindingStore
|
||
KnowledgeBases KnowledgeBaseLister
|
||
// ProductionBindings is the 源库 → 目标库 mapping store; ProductionKnowledgeBases
|
||
// enumerates the production-side bases the operator can bind.
|
||
// BackupSchedule 是「系统设置 → 存储与备份」的计划存储；BackupHost 负责在目标库
|
||
// 所在主机上读回/下发计划并触发一次整库备份（未接线时页面降级展示）。
|
||
// VersionHistory 是「版本记录」页的只读查询面；VersionHistoryRepo 是固定的仓库标识
|
||
// （owner/name）。两者都给才接线，否则页面降级为"未接线"（不报 500）。
|
||
VersionHistory VersionHistoryStore
|
||
VersionHistoryRepo string
|
||
BackupSchedule BackupScheduleStore
|
||
BackupHost BackupHost
|
||
BackupRuns *BackupRunTracker
|
||
ProductionBindings ProductionBindingStore
|
||
ProductionKnowledgeBases ProductionKnowledgeBaseLister
|
||
ProductionWeKnoraBaseURL string
|
||
ServiceTargetVerifier ServiceTargetVerifier
|
||
Now func() time.Time
|
||
RequireWeKnoraReady bool
|
||
ProductionReleaseRequestsEnabled bool
|
||
// TestAutoReviewPolicy 只用于展示：让运维在界面上看到"现在自动批准到什么程度"。
|
||
TestAutoReviewPolicy string
|
||
ProductionBackupRestoreVerified bool
|
||
ProductionReleaseAuthorizerRef string
|
||
ProductionReleaseAuthorizationToken string
|
||
// DingtalkSources 枚举「我账号下能看到」的钉钉知识库与钉盘空间（只读，dws）。
|
||
// 未接线时对应路由返回 501，页面显示"列不出来"而不是空列表。
|
||
DingtalkSources dingtalkSourcesLister
|
||
NewID func() (string, error)
|
||
Auth *AuthConfig
|
||
}
|
||
|
||
func NewHandler(config Config) (http.Handler, error) {
|
||
if config.Store == nil {
|
||
return nil, errors.New("job store is required")
|
||
}
|
||
if config.Now == nil {
|
||
config.Now = time.Now
|
||
}
|
||
if config.Monitor == nil {
|
||
config.Monitor = monitoring.NewSimulationProvider(config.Now)
|
||
}
|
||
if config.BackupRuns == nil {
|
||
config.BackupRuns = &BackupRunTracker{}
|
||
}
|
||
if config.BackupHost == nil {
|
||
// 未挂载主机凭据：Available()=false，页面显示"未接线"而不是 500。
|
||
config.BackupHost = backuphost.New(backuphost.Config{})
|
||
}
|
||
for _, knowledgeBaseID := range []string{config.InternalRoutingKnowledgeBaseID, config.ExternalRoutingKnowledgeBaseID} {
|
||
if knowledgeBaseID != "" && !validCatalogID(knowledgeBaseID) {
|
||
return nil, errors.New("routing knowledge base ID is invalid")
|
||
}
|
||
}
|
||
if config.ProductionReleaseRequestsEnabled &&
|
||
(strings.TrimSpace(config.ProductionReleaseAuthorizerRef) == "" || strings.TrimSpace(config.ProductionReleaseAuthorizationToken) == "") {
|
||
return nil, errors.New("production release authorization identity and token are required")
|
||
}
|
||
if config.NewID == nil {
|
||
config.NewID = newID
|
||
}
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("GET /api/v1/qa-ops/organization", func(w http.ResponseWriter, r *http.Request) { qaOrganization(w, r, config) })
|
||
mux.HandleFunc("POST /api/v1/qa-ops/organization/preview", func(w http.ResponseWriter, r *http.Request) { qaOrganization(w, r, config) })
|
||
mux.HandleFunc("PUT /api/v1/qa-ops/organization/regions", func(w http.ResponseWriter, r *http.Request) { qaOrganization(w, r, config) })
|
||
mux.HandleFunc("GET /api/v1/qa-ops/status", func(w http.ResponseWriter, r *http.Request) {
|
||
qaOpsStatus(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/qa-ops/records", func(w http.ResponseWriter, r *http.Request) {
|
||
listQAOpsRecords(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/qa-ops/records/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getQAOpsRecord(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
|
||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||
})
|
||
// P10: liveness answers "is this process serving HTTP" and NEVER depends on
|
||
// third parties; readiness answers "can this instance accept dependent work".
|
||
mux.HandleFunc("GET /livez", func(w http.ResponseWriter, _ *http.Request) {
|
||
writeJSON(w, http.StatusOK, map[string]any{"status": "live", "observed_at": config.Now().UTC()})
|
||
})
|
||
mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) {
|
||
health := config.Monitor.Services(r.Context())
|
||
// The API's own hard dependencies gate readiness. Worker and MinerU
|
||
// outages degrade the matrix (visible in Services) but must not take
|
||
// the admin UI out of rotation — the page is how operators fix them.
|
||
critical := map[string]bool{"postgres": true, "queue": true}
|
||
blocked := make([]string, 0)
|
||
degraded := make([]string, 0)
|
||
warnings := make([]string, 0)
|
||
for _, item := range health {
|
||
if !item.Applicable {
|
||
continue
|
||
}
|
||
if !item.Configured {
|
||
warnings = append(warnings, item.Key)
|
||
continue
|
||
}
|
||
key := strings.SplitN(item.Key, ":", 2)[0]
|
||
if item.Readiness == "ready" {
|
||
continue
|
||
}
|
||
if critical[key] || key == "weknora" && config.RequireWeKnoraReady {
|
||
blocked = append(blocked, item.Key)
|
||
} else {
|
||
degraded = append(degraded, item.Key)
|
||
}
|
||
}
|
||
status := map[string]any{"status": "ready", "blocked": blocked, "degraded": degraded, "not_configured": warnings, "observed_at": config.Now().UTC()}
|
||
if len(blocked) > 0 {
|
||
status["status"] = "blocked"
|
||
writeJSON(w, http.StatusServiceUnavailable, status)
|
||
return
|
||
}
|
||
if len(degraded) > 0 {
|
||
status["status"] = "degraded"
|
||
}
|
||
writeJSON(w, http.StatusOK, status)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/jobs", func(w http.ResponseWriter, r *http.Request) {
|
||
createJob(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/jobs", func(w http.ResponseWriter, r *http.Request) {
|
||
listJobs(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-items", func(w http.ResponseWriter, r *http.Request) {
|
||
listSourceAssets(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/sources", func(w http.ResponseWriter, r *http.Request) {
|
||
listCatalogSources(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/nodes", func(w http.ResponseWriter, r *http.Request) {
|
||
listCatalogNodes(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/snapshots", func(w http.ResponseWriter, r *http.Request) {
|
||
listCatalogSnapshots(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/snapshots/{id}/nodes", func(w http.ResponseWriter, r *http.Request) {
|
||
listCatalogSnapshotNodes(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/change-sets/latest", func(w http.ResponseWriter, r *http.Request) {
|
||
getLatestCatalogChangeSet(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-catalog/routing-capabilities", func(w http.ResponseWriter, r *http.Request) {
|
||
internal, external := config.effectiveRoutingTargets(r.Context())
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": map[string]bool{"internal_enabled": internal != "", "external_enabled": external != ""}})
|
||
})
|
||
mux.HandleFunc("GET /api/v1/routing-bindings", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.RoutingBindings == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "routing bindings storage unavailable"})
|
||
return
|
||
}
|
||
config.handleRoutingBindings(w, r)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/routing-bindings/options", func(w http.ResponseWriter, r *http.Request) {
|
||
config.handleRoutingKnowledgeBaseOptions(w, r)
|
||
})
|
||
mux.HandleFunc("PUT /api/v1/routing-bindings", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.RoutingBindings == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "routing bindings storage unavailable"})
|
||
return
|
||
}
|
||
config.handleSaveRoutingBindings(w, r)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/backup-schedule", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.BackupSchedule == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "backup schedule storage unavailable"})
|
||
return
|
||
}
|
||
config.handleBackupSchedule(w, r)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/backup-schedule/host", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.BackupSchedule == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "backup schedule storage unavailable"})
|
||
return
|
||
}
|
||
config.handleBackupScheduleHost(w, r)
|
||
})
|
||
mux.HandleFunc("PUT /api/v1/backup-schedule", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.BackupSchedule == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "backup schedule storage unavailable"})
|
||
return
|
||
}
|
||
config.handleSaveBackupSchedule(w, r)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/backup-schedule/apply", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.BackupSchedule == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "backup schedule storage unavailable"})
|
||
return
|
||
}
|
||
config.handleApplyBackupSchedule(w, r)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/backup-schedule/run", func(w http.ResponseWriter, r *http.Request) {
|
||
if config.BackupSchedule == nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "backup schedule storage unavailable"})
|
||
return
|
||
}
|
||
config.handleRunBackupNow(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)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-catalog/sources/{id}/refresh", func(w http.ResponseWriter, r *http.Request) {
|
||
triggerCatalogRefresh(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-catalog/change-sets/{id}/confirm", func(w http.ResponseWriter, r *http.Request) {
|
||
confirmCatalogChangeSet(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/monitor/overview", func(w http.ResponseWriter, r *http.Request) {
|
||
monitorOverview(w, r, config.Monitor)
|
||
})
|
||
// 「版本记录」：当前运行的构建身份（只读，与监控同级可见）
|
||
mux.HandleFunc("GET /api/v1/build-info", func(w http.ResponseWriter, r *http.Request) {
|
||
buildInfo(w, r)
|
||
})
|
||
// 全部 main 提交时间线（读本库缓存，不直接打 Gitea）
|
||
mux.HandleFunc("GET /api/v1/version-history", func(w http.ResponseWriter, r *http.Request) {
|
||
versionHistoryList(w, r, config)
|
||
})
|
||
// 手动同步一次（仅管理员；并发同步合并成同一个作业）
|
||
mux.HandleFunc("POST /api/v1/version-history/sync", func(w http.ResponseWriter, r *http.Request) {
|
||
versionHistorySync(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/monitor/knowledge-bases/{id}/files", func(w http.ResponseWriter, r *http.Request) {
|
||
monitorFiles(w, r, config.Monitor)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/monitor/metrics", func(w http.ResponseWriter, r *http.Request) {
|
||
monitorMetrics(w, r, config.Monitor)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/scope-rules", func(w http.ResponseWriter, r *http.Request) {
|
||
listScopeRules(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("PUT /api/v1/scope-rules", func(w http.ResponseWriter, r *http.Request) {
|
||
mutateScopeRule(w, r, config, false)
|
||
})
|
||
mux.HandleFunc("DELETE /api/v1/scope-rules", func(w http.ResponseWriter, r *http.Request) {
|
||
mutateScopeRule(w, r, config, true)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/filename-filters", func(w http.ResponseWriter, r *http.Request) {
|
||
listFilenameFilters(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("PUT /api/v1/filename-filters", func(w http.ResponseWriter, r *http.Request) {
|
||
mutateFilenameFilter(w, r, config, false)
|
||
})
|
||
mux.HandleFunc("DELETE /api/v1/filename-filters", func(w http.ResponseWriter, r *http.Request) {
|
||
mutateFilenameFilter(w, r, config, true)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-connections", func(w http.ResponseWriter, r *http.Request) {
|
||
listSourceConnections(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/dingtalk-session", func(w http.ResponseWriter, r *http.Request) {
|
||
startDingtalkSession(w, r)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/dingtalk-spaces", func(w http.ResponseWriter, r *http.Request) {
|
||
registerDingtalkSpace(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/dingtalk-spaces/available", func(w http.ResponseWriter, r *http.Request) {
|
||
listDingtalkSources(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/dingtalk-session/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getDingtalkSession(w, r)
|
||
})
|
||
mux.HandleFunc("DELETE /api/v1/dingtalk-session/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
deleteDingtalkSession(w, r)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-connections/{id}/scan", func(w http.ResponseWriter, r *http.Request) {
|
||
triggerPPSourceScan(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-connections/{id}/scans", func(w http.ResponseWriter, r *http.Request) {
|
||
listSourceScans(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-connections/{id}/scans/{scan_id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getSourceScanDetail(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/source-connections/{id}/settings/revisions", func(w http.ResponseWriter, r *http.Request) {
|
||
listSourceSettingsRevisions(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-connections/{id}/settings/revisions", func(w http.ResponseWriter, r *http.Request) {
|
||
createSourceSettingsRevision(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-connections/{id}/settings/revisions/{revision_id}/activate", func(w http.ResponseWriter, r *http.Request) {
|
||
activateSourceSettingsRevision(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-connections/{id}/archive", func(w http.ResponseWriter, r *http.Request) {
|
||
archiveSourceConnection(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/source-connections/{id}/restore", func(w http.ResponseWriter, r *http.Request) {
|
||
restoreSourceConnection(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/review-candidates", func(w http.ResponseWriter, r *http.Request) {
|
||
listReviewCandidates(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/review-candidates/batch-decision", func(w http.ResponseWriter, r *http.Request) {
|
||
batchReviewCandidates(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/review-candidates/{id}/decision", func(w http.ResponseWriter, r *http.Request) {
|
||
reviewPublication(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-batches", func(w http.ResponseWriter, r *http.Request) {
|
||
listReleaseBatches(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-executions/{id}/audit-chain", func(w http.ResponseWriter, r *http.Request) {
|
||
getReleaseAuditChain(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-change-sets", func(w http.ResponseWriter, r *http.Request) {
|
||
generateReleaseChangeSet(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-change-sets", func(w http.ResponseWriter, r *http.Request) {
|
||
listReleaseChangeSets(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-change-sets/latest", func(w http.ResponseWriter, r *http.Request) {
|
||
getReleaseChangeSet(w, r, config.Store, "latest")
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-change-sets/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getReleaseChangeSet(w, r, config.Store, r.PathValue("id"))
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-change-sets/{id}/confirm", func(w http.ResponseWriter, r *http.Request) {
|
||
confirmReleaseChangeSet(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-plan", func(w http.ResponseWriter, r *http.Request) {
|
||
plan, err := config.Store.ListReleasePlan(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not build release plan"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": plan})
|
||
})
|
||
mux.HandleFunc("GET /api/v1/ui-settings", func(w http.ResponseWriter, r *http.Request) {
|
||
getActiveUISettings(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/ui-settings/candidates", func(w http.ResponseWriter, r *http.Request) {
|
||
createUISettingsCandidate(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/ui-settings/{id}/activate", func(w http.ResponseWriter, r *http.Request) {
|
||
activateUISettings(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-batches", func(w http.ResponseWriter, r *http.Request) {
|
||
createReleaseBatch(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-batches/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getReleaseBatch(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-batches/{id}/seal", func(w http.ResponseWriter, r *http.Request) {
|
||
sealReleaseBatch(w, r, config)
|
||
})
|
||
// 一个动作把"建批（冻结+锁定）→ 授权生产发布"走完：门禁、固定确认语句、管理员身份、
|
||
// 发布前快照门禁全部沿用既有实现，不新增旁路。
|
||
mux.HandleFunc("POST /api/v1/release-batches/publish-now", func(w http.ResponseWriter, r *http.Request) {
|
||
publishNowPublications(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-batches/{id}/authorize-production", func(w http.ResponseWriter, r *http.Request) {
|
||
authorizeProductionRelease(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-executions/{id}/authorize-rollback", func(w http.ResponseWriter, r *http.Request) {
|
||
authorizeProductionRollback(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/release-executions/{id}/retry", func(w http.ResponseWriter, r *http.Request) {
|
||
retryProductionRelease(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-executions", func(w http.ResponseWriter, r *http.Request) {
|
||
listReleaseExecutions(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-executions/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
getReleaseExecution(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/release-capabilities", func(w http.ResponseWriter, _ *http.Request) {
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": map[string]any{
|
||
"production_release_requests_enabled": config.ProductionReleaseRequestsEnabled,
|
||
"production_backup_restore_verified": config.ProductionBackupRestoreVerified,
|
||
"test_auto_review_policy": config.TestAutoReviewPolicy,
|
||
}})
|
||
})
|
||
mux.HandleFunc("GET /api/v1/test-publications", func(w http.ResponseWriter, r *http.Request) {
|
||
listTestPublications(w, r, config.Store)
|
||
})
|
||
// 需下架条目的移除入口：把「被禁止导入但仍存活在测试库」的知识从测试库移除
|
||
// （置 disabling + 入队既有 test.remove 工作流）。与开始处理同级：engineer 即可。
|
||
mux.HandleFunc("POST /api/v1/test-publications/{id}/removal", func(w http.ResponseWriter, r *http.Request) {
|
||
startTestPublicationRemoval(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/test-publications/start", func(w http.ResponseWriter, r *http.Request) {
|
||
startTestPublications(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/test-publications/{id}/acceptance", func(w http.ResponseWriter, r *http.Request) {
|
||
listAcceptance(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/test-publications/{id}/retry", func(w http.ResponseWriter, r *http.Request) {
|
||
retryPublication(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/test-publications/retry-batch", func(w http.ResponseWriter, r *http.Request) {
|
||
retryPublicationsBatch(w, r, config)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/test-publications/cancel-batch", func(w http.ResponseWriter, r *http.Request) {
|
||
cancelPublicationsBatch(w, r, config)
|
||
})
|
||
mux.HandleFunc("GET /api/v1/job-failures", func(w http.ResponseWriter, r *http.Request) {
|
||
listJobFailures(w, r, config.Store)
|
||
})
|
||
mux.HandleFunc("POST /api/v1/job-failures/{id}/requeue", func(w http.ResponseWriter, r *http.Request) {
|
||
requeueJobFailure(w, r, config)
|
||
})
|
||
if config.Auth != nil {
|
||
if config.Auth.Store == nil {
|
||
return nil, errors.New("auth configuration requires a store")
|
||
}
|
||
config.authRoutes(mux)
|
||
config.userRoutes(mux)
|
||
return config.Auth.authenticate(mux), nil
|
||
}
|
||
return mux, nil
|
||
}
|
||
|
||
type createJobRequest struct {
|
||
Type string `json:"type"`
|
||
Payload json.RawMessage `json:"payload"`
|
||
Priority int `json:"priority"`
|
||
MaxAttempts int `json:"max_attempts"`
|
||
IdempotencyKey string `json:"idempotency_key"`
|
||
}
|
||
|
||
func createJob(w http.ResponseWriter, r *http.Request, config Config) {
|
||
var input createJobRequest
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<20))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(&input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON request"})
|
||
return
|
||
}
|
||
if strings.TrimSpace(input.Type) == "" || len(input.Type) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "type is required and must not exceed 100 characters"})
|
||
return
|
||
}
|
||
if strings.TrimSpace(input.IdempotencyKey) == "" || len(input.IdempotencyKey) > 255 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "idempotency_key is required and must not exceed 255 characters"})
|
||
return
|
||
}
|
||
if input.MaxAttempts == 0 {
|
||
input.MaxAttempts = 5
|
||
}
|
||
if input.MaxAttempts < 1 || input.MaxAttempts > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "max_attempts must be between 1 and 100"})
|
||
return
|
||
}
|
||
if len(input.Payload) == 0 {
|
||
input.Payload = json.RawMessage(`{}`)
|
||
}
|
||
if !json.Valid(input.Payload) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "payload must be valid JSON"})
|
||
return
|
||
}
|
||
id, err := config.NewID()
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not create job ID"})
|
||
return
|
||
}
|
||
now := config.Now().UTC()
|
||
job, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
|
||
ID: id,
|
||
Type: input.Type,
|
||
Payload: input.Payload,
|
||
Priority: input.Priority,
|
||
AvailableAt: now,
|
||
MaxAttempts: input.MaxAttempts,
|
||
IdempotencyKey: input.IdempotencyKey,
|
||
}, now)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not enqueue job"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, map[string]any{"data": job})
|
||
}
|
||
|
||
func listJobs(w http.ResponseWriter, r *http.Request, store JobStore) {
|
||
limit := 50
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
limit = value
|
||
}
|
||
result, err := store.List(r.Context(), limit)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list jobs"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result})
|
||
}
|
||
|
||
func listSourceAssets(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.SourceAssetFilter{State: r.URL.Query().Get("state"), Limit: 50}
|
||
if filter.State != "" && filter.State != "completed" && filter.State != "content_alias" && filter.State != "over_limit" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "state must be completed, content_alias, or over_limit"})
|
||
return
|
||
}
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
filter.Limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
filter.Offset = value
|
||
}
|
||
page, err := store.ListSourceAssets(r.Context(), filter)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source items"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{
|
||
"data": page.Items, "summary": page.Summary, "limit": filter.Limit, "offset": filter.Offset,
|
||
})
|
||
}
|
||
|
||
func listCatalogSources(w http.ResponseWriter, r *http.Request, store Store) {
|
||
sources, err := store.ListCatalogSources(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source catalogs"})
|
||
return
|
||
}
|
||
sources = append(sources, scanningCatalogSources(r.Context(), store)...)
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": sources, "summary": map[string]int{"total": len(sources)}})
|
||
}
|
||
|
||
// catalogScanProgressLister 是本路由需要的最小能力面：列出「已登记、还没有快照」的来源的
|
||
// 首次扫描进度。真实 Store 实现它；未接线时安静跳过（只是没有"扫描中"那几行，不影响既有列表）。
|
||
type catalogScanProgressLister interface {
|
||
ListCatalogScanProgress(context.Context) ([]governance.CatalogScanProgress, error)
|
||
}
|
||
|
||
// scanningCatalogSources 把首次扫描中的来源投影成目录列表里的行：新来源不必等快照落地才出现。
|
||
func scanningCatalogSources(ctx context.Context, store Store) []governance.CatalogSource {
|
||
lister, ok := store.(catalogScanProgressLister)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
progress, err := lister.ListCatalogScanProgress(ctx)
|
||
if err != nil {
|
||
return nil
|
||
}
|
||
out := make([]governance.CatalogSource, 0, len(progress))
|
||
for _, item := range progress {
|
||
out = append(out, governance.CatalogSource{
|
||
ConnectionID: item.ConnectionID,
|
||
SourceType: item.SourceType,
|
||
Name: item.Name,
|
||
// 还没有快照：根节点与计数只报"已扫到多少"，由前端按"首次扫描中"呈现。
|
||
RootNodeID: "",
|
||
FolderCount: item.DiscoveredDirectories,
|
||
FileCount: item.ScannedFiles,
|
||
ObservedAt: item.StartedAt,
|
||
SnapshotOnly: true,
|
||
SnapshotStatus: "scanning",
|
||
ScanProgress: item.Progress,
|
||
ScanMessage: item.Message,
|
||
})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func listCatalogNodes(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.CatalogNodeFilter{ConnectionID: r.URL.Query().Get("connection_id"), ParentNodeID: r.URL.Query().Get("parent_id"), Limit: 50}
|
||
if !validCatalogID(filter.ConnectionID) || !validCatalogID(filter.ParentNodeID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source catalog location"})
|
||
return
|
||
}
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
filter.Limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
filter.Offset = value
|
||
}
|
||
page, err := store.ListCatalogNodes(r.Context(), filter)
|
||
if errors.Is(err, postgresstore.ErrCatalogNodeNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source catalog location not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source catalog"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
func listCatalogSnapshots(w http.ResponseWriter, r *http.Request, store Store) {
|
||
connectionID := r.URL.Query().Get("connection_id")
|
||
if !validCatalogID(connectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source catalog"})
|
||
return
|
||
}
|
||
limit, offset, ok := parseLimitOffset(w, r, 20)
|
||
if !ok {
|
||
return
|
||
}
|
||
page, err := store.ListCatalogSnapshots(r.Context(), connectionID, limit, offset)
|
||
if errors.Is(err, postgresstore.ErrCatalogSnapshotNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source catalog not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list snapshot history"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
func listCatalogSnapshotNodes(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.CatalogSnapshotNodeFilter{ConnectionID: r.URL.Query().Get("connection_id"), SnapshotID: r.PathValue("id"), ParentNodeID: r.URL.Query().Get("parent_id"), Limit: 50}
|
||
if !validCatalogID(filter.ConnectionID) || !validCatalogID(filter.SnapshotID) || !validCatalogID(filter.ParentNodeID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid snapshot location"})
|
||
return
|
||
}
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
filter.Limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
filter.Offset = value
|
||
}
|
||
page, err := store.ListCatalogNodesInSnapshot(r.Context(), filter)
|
||
if errors.Is(err, postgresstore.ErrCatalogSnapshotNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "snapshot not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list snapshot nodes"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
func getLatestCatalogChangeSet(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.CatalogChangeSetFilter{ConnectionID: strings.TrimSpace(r.URL.Query().Get("connection_id")), Action: strings.TrimSpace(r.URL.Query().Get("action")), Limit: 20}
|
||
if !validCatalogID(filter.ConnectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source catalog"})
|
||
return
|
||
}
|
||
allowedActions := map[string]bool{"": true, "add": true, "update": true, "move": true, "move_scope": true, "delete": true, "no_change": true}
|
||
if !allowedActions[filter.Action] {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid change action"})
|
||
return
|
||
}
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
filter.Limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
filter.Offset = value
|
||
}
|
||
detail, err := store.GetLatestCatalogChangeSet(r.Context(), filter)
|
||
if errors.Is(err, postgresstore.ErrCatalogChangeSetNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source catalog change set not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load source catalog change set"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": detail})
|
||
}
|
||
|
||
func validCatalogID(value string) bool {
|
||
if value == "" || len(value) > 36 {
|
||
return false
|
||
}
|
||
for _, character := range value {
|
||
if character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z' || character >= '0' && character <= '9' || character == '-' || character == '_' {
|
||
continue
|
||
}
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
|
||
func requireOperationID(w http.ResponseWriter, r *http.Request) (string, bool) {
|
||
operationID := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
|
||
if operationID == "" || len(operationID) > 128 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "Idempotency-Key is required and must not exceed 128 characters"})
|
||
return "", false
|
||
}
|
||
return operationID, true
|
||
}
|
||
|
||
type catalogRouteInput struct {
|
||
ConnectionID string `json:"connection_id"`
|
||
NodeIDs []string `json:"node_ids"`
|
||
Policy string `json:"policy"`
|
||
Visibility string `json:"visibility"`
|
||
Reason string `json:"reason"`
|
||
}
|
||
|
||
func routeCatalogFiles(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input catalogRouteInput
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<16))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(&input); err != nil || len(input.NodeIDs) == 0 || len(input.NodeIDs) > 100 || !validCatalogID(input.ConnectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid catalog route request"})
|
||
return
|
||
}
|
||
policy := strings.TrimSpace(input.Policy)
|
||
if policy == "" {
|
||
policy = strings.TrimSpace(input.Visibility)
|
||
}
|
||
visibility := governance.Visibility(policy)
|
||
effectiveInternal, effectiveExternal := config.effectiveRoutingTargets(r.Context())
|
||
targetKnowledgeBaseIDs := make([]string, 0, 2)
|
||
switch visibility {
|
||
case governance.VisibilityInternal:
|
||
if effectiveInternal != "" {
|
||
targetKnowledgeBaseIDs = append(targetKnowledgeBaseIDs, effectiveInternal)
|
||
}
|
||
case governance.VisibilityExternal:
|
||
if effectiveExternal != "" {
|
||
targetKnowledgeBaseIDs = append(targetKnowledgeBaseIDs, effectiveExternal)
|
||
}
|
||
case governance.VisibilityBoth:
|
||
if effectiveInternal != "" {
|
||
targetKnowledgeBaseIDs = append(targetKnowledgeBaseIDs, effectiveInternal)
|
||
}
|
||
if effectiveExternal != "" {
|
||
targetKnowledgeBaseIDs = append(targetKnowledgeBaseIDs, effectiveExternal)
|
||
}
|
||
case governance.VisibilityBlocked:
|
||
default:
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "policy must be internal, external, both, or blocked"})
|
||
return
|
||
}
|
||
if visibility != governance.VisibilityBlocked &&
|
||
((visibility == governance.VisibilityBoth && len(targetKnowledgeBaseIDs) != 2) ||
|
||
(visibility != governance.VisibilityBoth && len(targetKnowledgeBaseIDs) != 1)) {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "routing target is not configured"})
|
||
return
|
||
}
|
||
targetKnowledgeBaseID := ""
|
||
if len(targetKnowledgeBaseIDs) > 0 {
|
||
targetKnowledgeBaseID = targetKnowledgeBaseIDs[0]
|
||
}
|
||
result, replayed, err := config.Store.RouteCatalogFilesWithOperation(r.Context(), governance.CatalogRouteRequest{
|
||
ConnectionID: input.ConnectionID, NodeIDs: input.NodeIDs, Visibility: visibility,
|
||
TargetKnowledgeBaseID: targetKnowledgeBaseID, TargetKnowledgeBaseIDs: targetKnowledgeBaseIDs,
|
||
OperatorRef: actorFor(r), Reason: strings.TrimSpace(input.Reason), Now: config.Now().UTC(),
|
||
}, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "could not route selected files"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
|
||
func triggerCatalogRefresh(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
connectionID := r.PathValue("id")
|
||
if !validCatalogID(connectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source catalog"})
|
||
return
|
||
}
|
||
sources, err := config.Store.ListCatalogSources(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not inspect source catalog"})
|
||
return
|
||
}
|
||
var source *governance.CatalogSource
|
||
for index := range sources {
|
||
if sources[index].ConnectionID == connectionID {
|
||
source = &sources[index]
|
||
break
|
||
}
|
||
}
|
||
if source == nil {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source catalog not found"})
|
||
return
|
||
}
|
||
job, replayed, err := config.Store.EnqueueCatalogRefreshWithOperation(r.Context(), *source, operationID, config.Now().UTC())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not enqueue catalog refresh"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": job, "operation": operation})
|
||
}
|
||
|
||
func monitorOverview(w http.ResponseWriter, r *http.Request, provider MonitorProvider) {
|
||
overview, err := provider.Overview(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load monitor overview"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, overview)
|
||
}
|
||
|
||
func monitorMetrics(w http.ResponseWriter, r *http.Request, provider MonitorProvider) {
|
||
metrics, err := provider.Metrics(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "run metrics unavailable"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": metrics})
|
||
}
|
||
|
||
func monitorFiles(w http.ResponseWriter, r *http.Request, provider MonitorProvider) {
|
||
knowledgeBaseID := r.PathValue("id")
|
||
status := r.URL.Query().Get("status")
|
||
allowedStatus := map[string]bool{"": true, "completed": true, "pending": true, "processing": true, "finalizing": true, "failed": true, "cancelled": true}
|
||
if !validCatalogID(knowledgeBaseID) || !allowedStatus[status] {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid monitor file query"})
|
||
return
|
||
}
|
||
limit, offset := 20, 0
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
offset = value
|
||
}
|
||
page, err := provider.ListFiles(r.Context(), knowledgeBaseID, status, limit, offset)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load monitored files"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
type scopeRuleInput struct {
|
||
TargetType governance.ScopeTargetType `json:"target_type"`
|
||
ConnectionID string `json:"connection_id"`
|
||
DirectoryPath string `json:"directory_path"`
|
||
SourceItemID string `json:"source_item_id"`
|
||
Visibility governance.Visibility `json:"visibility"`
|
||
|
||
Reason string `json:"reason"`
|
||
}
|
||
|
||
func listScopeRules(w http.ResponseWriter, r *http.Request, store Store) {
|
||
connectionID := strings.TrimSpace(r.URL.Query().Get("connection_id"))
|
||
if connectionID == "" || len(connectionID) > 36 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "connection_id is required"})
|
||
return
|
||
}
|
||
rules, err := store.ListScopeRules(r.Context(), connectionID)
|
||
if errors.Is(err, postgresstore.ErrScopeTargetNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "scope target not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list scope rules"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": rules, "summary": map[string]int{"total": len(rules)}})
|
||
}
|
||
|
||
func mutateScopeRule(w http.ResponseWriter, r *http.Request, config Config, clear bool) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input scopeRuleInput
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<16))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(&input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid scope rule request"})
|
||
return
|
||
}
|
||
request := governance.ScopeRuleMutation{
|
||
TargetType: input.TargetType, ConnectionID: strings.TrimSpace(input.ConnectionID),
|
||
DirectoryPath: input.DirectoryPath, SourceItemID: strings.TrimSpace(input.SourceItemID),
|
||
Visibility: input.Visibility, OperatorRef: actorFor(r), Reason: strings.TrimSpace(input.Reason),
|
||
Clear: clear, Now: config.Now().UTC(),
|
||
}
|
||
if clear {
|
||
request.Visibility = ""
|
||
}
|
||
if err := request.Validate(); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid scope rule request"})
|
||
return
|
||
}
|
||
rule, replayed, err := config.Store.MutateScopeRuleWithOperation(r.Context(), request, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrScopeTargetNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "scope target not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not update scope rule"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": rule, "operation": operation})
|
||
}
|
||
|
||
type sourceSettingsRevisionRequest struct {
|
||
SourceType string `json:"source_type"`
|
||
Name string `json:"name"`
|
||
BaseURL string `json:"endpoint"`
|
||
SecretRef string `json:"credential_ref"`
|
||
Enabled bool `json:"enabled"`
|
||
Schedule struct {
|
||
RefreshIntervalSeconds *int `json:"refresh_interval_seconds"`
|
||
} `json:"schedule"`
|
||
}
|
||
|
||
// toDraft stamps the actor server-side; the browser never names the operator.
|
||
func (req sourceSettingsRevisionRequest) toDraft(actor string) postgresstore.SourceSettingsDraft {
|
||
schedule := "{}"
|
||
if req.Schedule.RefreshIntervalSeconds != nil {
|
||
schedule = fmt.Sprintf(`{"refresh_interval_seconds":%d}`, *req.Schedule.RefreshIntervalSeconds)
|
||
}
|
||
return postgresstore.SourceSettingsDraft{
|
||
SourceType: req.SourceType, Name: req.Name, BaseURL: req.BaseURL,
|
||
SecretRef: req.SecretRef, Enabled: req.Enabled,
|
||
Schedule: json.RawMessage(schedule), Operator: actor,
|
||
}
|
||
}
|
||
|
||
func listSourceSettingsRevisions(w http.ResponseWriter, r *http.Request, store Store) {
|
||
connectionID := r.PathValue("id")
|
||
if !validCatalogID(connectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source connection"})
|
||
return
|
||
}
|
||
revisions, err := store.ListSourceConnectionRevisions(r.Context(), connectionID)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source settings revisions"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": revisions})
|
||
}
|
||
|
||
func createSourceSettingsRevision(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if r.Header.Get("X-Governance-Action") != "source-settings" {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "source settings write intent is required"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
connectionID := r.PathValue("id")
|
||
if !validCatalogID(connectionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source connection"})
|
||
return
|
||
}
|
||
var request sourceSettingsRevisionRequest
|
||
if err := decodeJSONStrict(r, &request); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source settings revision"})
|
||
return
|
||
}
|
||
revisionID, replayed, err := config.Store.CreateSourceSettingsRevisionWithOperation(r.Context(), connectionID, request.toDraft(actorFor(r)), operationID, config.Now().UTC())
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrInvalidSourceSettings) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source settings revision"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrSourceConnectionNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source connection not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not create source settings revision"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusCreated, map[string]any{"data": map[string]string{"id": revisionID}, "operation": operation})
|
||
}
|
||
|
||
func activateSourceSettingsRevision(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if r.Header.Get("X-Governance-Action") != "source-settings" {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "source settings write intent is required"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
connectionID, revisionID := r.PathValue("id"), r.PathValue("revision_id")
|
||
if !validCatalogID(connectionID) || !validCatalogID(revisionID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source settings revision"})
|
||
return
|
||
}
|
||
var request struct{}
|
||
if err := decodeJSONStrict(r, &request); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source settings activation"})
|
||
return
|
||
}
|
||
replayed, err := config.Store.ActivateSourceSettingsRevisionWithOperation(r.Context(), connectionID, revisionID, operationID, config.Now().UTC())
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrInvalidSourceSettings) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "only validated revisions can be activated"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrSourceConnectionNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source settings revision not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not activate source settings revision"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": map[string]string{"connection_id": connectionID, "revision_id": revisionID}, "operation": operation})
|
||
}
|
||
|
||
func listSourceConnections(w http.ResponseWriter, r *http.Request, store Store) {
|
||
page, err := store.ListSourceConnections(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source connections"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page.Items, "summary": page.Summary})
|
||
}
|
||
|
||
func triggerPPSourceScan(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
scan, replayed, err := config.Store.TriggerPPSourceScanWithOperation(r.Context(), r.PathValue("id"), config.Now().UTC(), operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrSourceConnectionNotScannable) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source connection is not scannable"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrSourceScanAlreadyActive) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "source scan already active"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not trigger source scan"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": scan, "operation": operation})
|
||
}
|
||
|
||
func listSourceScans(w http.ResponseWriter, r *http.Request, store Store) {
|
||
connectionID := r.PathValue("id")
|
||
if connectionID == "" || len(connectionID) > 64 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source connection ID"})
|
||
return
|
||
}
|
||
limit, offset := 20, 0
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
offset = value
|
||
}
|
||
page, err := store.ListSourceScans(r.Context(), connectionID, limit, offset)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list source scans"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page.Items, "total": page.Total, "limit": limit, "offset": offset})
|
||
}
|
||
|
||
func getSourceScanDetail(w http.ResponseWriter, r *http.Request, store Store) {
|
||
connectionID, scanID := r.PathValue("id"), r.PathValue("scan_id")
|
||
if connectionID == "" || len(connectionID) > 64 || scanID == "" || len(scanID) > 64 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid source scan ID"})
|
||
return
|
||
}
|
||
detail, err := store.GetSourceScanDetail(r.Context(), connectionID, scanID)
|
||
if errors.Is(err, postgresstore.ErrSourceScanNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "source scan not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load source scan"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": detail})
|
||
}
|
||
|
||
func listReviewCandidates(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.ReviewCandidateFilter{ReviewState: r.URL.Query().Get("review_state"), Limit: 20}
|
||
if filter.ReviewState != "" && filter.ReviewState != "pending" && filter.ReviewState != "approved" && filter.ReviewState != "rejected" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "review_state must be pending, approved, or rejected"})
|
||
return
|
||
}
|
||
limit, offset, ok := parseLimitOffset(w, r, 20)
|
||
if !ok {
|
||
return
|
||
}
|
||
filter.Limit, filter.Offset = limit, offset
|
||
page, err := store.ListReviewCandidates(r.Context(), filter)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list review candidates"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page.Items, "total": page.Total, "limit": limit, "offset": offset})
|
||
}
|
||
|
||
type reviewPublicationRequest struct {
|
||
Decision string `json:"decision"`
|
||
DecidedVisibility string `json:"decided_visibility"`
|
||
DecisionCode string `json:"decision_code"`
|
||
}
|
||
|
||
func reviewPublication(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input reviewPublicationRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid review request"})
|
||
return
|
||
}
|
||
review, replayed, err := config.Store.ReviewPublicationWithOperation(r.Context(), governance.ReviewPublicationRequest{
|
||
PublicationID: r.PathValue("id"), Decision: governance.ReviewState(input.Decision),
|
||
DecidedVisibility: governance.Visibility(input.DecidedVisibility), ReviewerRef: actorFor(r),
|
||
DecisionCode: input.DecisionCode, Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidPublicationReview):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid review request"})
|
||
case errors.Is(err, postgresstore.ErrPublicationReviewNotEligible):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "publication is not eligible for this review decision"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not record review decision"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusCreated, map[string]any{"data": review, "operation": operation})
|
||
}
|
||
}
|
||
|
||
type createUISettingsCandidateRequest struct {
|
||
DocumentRefreshSeconds int `json:"document_refresh_seconds"`
|
||
ResourceRefreshSeconds int `json:"resource_refresh_seconds"`
|
||
StalledAfterSeconds int `json:"stalled_after_seconds"`
|
||
FinalizingAfterSeconds int `json:"finalizing_after_seconds"`
|
||
IngestionConcurrency int `json:"ingestion_concurrency"`
|
||
TestWeKnoraBaseURL string `json:"test_weknora_base_url"`
|
||
ProductionWeKnoraBaseURL string `json:"production_weknora_base_url"`
|
||
ProductionTenantID string `json:"production_tenant_id"`
|
||
ProductionKnowledgeBaseID string `json:"production_knowledge_base_id"`
|
||
}
|
||
|
||
// adminAuthorized reports whether the caller may change operator-level
|
||
// configuration (endpoints, production IDs). With auth enabled the RBAC middleware
|
||
// has already put a session identity in the context; without auth (development
|
||
// socket only) there is no identity to check, matching productionAuthorizer's
|
||
// fallback rather than inventing a second rule.
|
||
func adminAuthorized(r *http.Request) bool {
|
||
identity, ok := sessionFrom(r.Context())
|
||
if !ok {
|
||
return true
|
||
}
|
||
return identity.Role == roleAdmin
|
||
}
|
||
|
||
// operatorServiceFields are the operator-level entries of 「服务配置」: the test
|
||
// and production endpoints, the production tenant and the production knowledge
|
||
// base. Engineers may still tune refresh windows; only an admin may repoint the
|
||
// console's environments or rewrite its production IDs.
|
||
type operatorServiceFields struct {
|
||
TestWeKnoraBaseURL string
|
||
ProductionWeKnoraBaseURL string
|
||
ProductionTenantID string
|
||
ProductionKnowledgeBaseID string
|
||
}
|
||
|
||
func operatorServiceFieldsOf(settings governance.UISettings) operatorServiceFields {
|
||
return operatorServiceFields{
|
||
TestWeKnoraBaseURL: settings.TestWeKnoraBaseURL,
|
||
ProductionWeKnoraBaseURL: settings.ProductionWeKnoraBaseURL,
|
||
ProductionTenantID: settings.ProductionTenantID,
|
||
ProductionKnowledgeBaseID: settings.ProductionKnowledgeBaseID,
|
||
}
|
||
}
|
||
|
||
// operatorChangeNeedsAdmin keeps the operator-level service configuration
|
||
// admin-only on both ends of the versioned settings flow: authoring a candidate,
|
||
// and activating a revision whose values differ from the active ones (otherwise
|
||
// an engineer could activate an admin-authored change).
|
||
func operatorChangeNeedsAdmin(w http.ResponseWriter, r *http.Request, current, incoming operatorServiceFields) bool {
|
||
if current == incoming || adminAuthorized(r) {
|
||
return false
|
||
}
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "service configuration changes require an admin"})
|
||
return true
|
||
}
|
||
|
||
func getActiveUISettings(w http.ResponseWriter, r *http.Request, store Store) {
|
||
settings, err := store.GetActiveUISettings(r.Context())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load active UI settings"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": settings})
|
||
}
|
||
|
||
// ServiceTargetVerifier performs a real read against the configured endpoints so
|
||
// that 「服务配置」cannot store an ID nobody ever checked. It decides no policy:
|
||
// it answers, and the caller refuses only a definitive negative answer.
|
||
type ServiceTargetVerifier interface {
|
||
VerifyServiceTarget(ctx context.Context, target governance.ServiceTarget) governance.ServiceTargetVerification
|
||
}
|
||
|
||
// verifyServiceTarget runs the configured verifier over the staged service
|
||
// fields. When the endpoint answered "not there" it writes the 400 itself and
|
||
// reports rejected=true, so the caller stops before touching the store.
|
||
func verifyServiceTarget(w http.ResponseWriter, r *http.Request, config Config, fields operatorServiceFields) (governance.ServiceTargetVerification, bool) {
|
||
if config.ServiceTargetVerifier == nil {
|
||
return governance.ServiceTargetVerification{}, false
|
||
}
|
||
verification := config.ServiceTargetVerifier.VerifyServiceTarget(r.Context(), governance.ServiceTarget{
|
||
TestBaseURL: fields.TestWeKnoraBaseURL,
|
||
ProductionBaseURL: fields.ProductionWeKnoraBaseURL,
|
||
ProductionTenantID: fields.ProductionTenantID,
|
||
ProductionKnowledgeBaseID: fields.ProductionKnowledgeBaseID,
|
||
})
|
||
if verification.Blocked {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": verification.Message, "verification": verification})
|
||
return verification, true
|
||
}
|
||
return verification, false
|
||
}
|
||
|
||
// uiSettingsWriteResponse keeps the write response shape stable across the paths
|
||
// that do and do not run a real verification (Status is empty when no verifier
|
||
// is configured).
|
||
func uiSettingsWriteResponse(settings governance.UISettings, operation string, verification governance.ServiceTargetVerification) map[string]any {
|
||
payload := map[string]any{"data": settings, "operation": operation}
|
||
if verification.Status != "" {
|
||
payload["verification"] = verification
|
||
}
|
||
return payload
|
||
}
|
||
|
||
func createUISettingsCandidate(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if r.Header.Get("X-Governance-Action") != "ui-settings" {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "UI settings write intent is required"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var request createUISettingsCandidateRequest
|
||
if err := decodeJSONStrict(r, &request); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid UI settings candidate"})
|
||
return
|
||
}
|
||
// The endpoints and the production IDs are operator-level config: engineers
|
||
// may tune refresh windows, only an admin may repoint the console at a
|
||
// different environment or rewrite its production IDs.
|
||
incoming := operatorServiceFields{
|
||
TestWeKnoraBaseURL: strings.TrimSpace(request.TestWeKnoraBaseURL),
|
||
ProductionWeKnoraBaseURL: strings.TrimSpace(request.ProductionWeKnoraBaseURL),
|
||
ProductionTenantID: strings.TrimSpace(request.ProductionTenantID),
|
||
ProductionKnowledgeBaseID: strings.TrimSpace(request.ProductionKnowledgeBaseID),
|
||
}
|
||
current := operatorServiceFields{}
|
||
if active, err := config.Store.GetActiveUISettings(r.Context()); err == nil {
|
||
current = operatorServiceFieldsOf(active)
|
||
}
|
||
if operatorChangeNeedsAdmin(w, r, current, incoming) {
|
||
return
|
||
}
|
||
// The four service fields are only worth storing if the endpoint really has
|
||
// them, so ask it now: a definitive "not there" refuses the save, a flaky or
|
||
// unconfigured endpoint only warns (an outage must not stop an operator from
|
||
// fixing the address).
|
||
verification, rejected := verifyServiceTarget(w, r, config, incoming)
|
||
if rejected {
|
||
return
|
||
}
|
||
settings, replayed, err := config.Store.CreateUISettingsCandidateWithOperation(r.Context(), governance.CreateUISettingsCandidateRequest{
|
||
DocumentRefreshSeconds: request.DocumentRefreshSeconds,
|
||
ResourceRefreshSeconds: request.ResourceRefreshSeconds,
|
||
StalledAfterSeconds: request.StalledAfterSeconds,
|
||
FinalizingAfterSeconds: request.FinalizingAfterSeconds,
|
||
IngestionConcurrency: request.IngestionConcurrency,
|
||
TestWeKnoraBaseURL: incoming.TestWeKnoraBaseURL,
|
||
ProductionWeKnoraBaseURL: incoming.ProductionWeKnoraBaseURL,
|
||
ProductionTenantID: incoming.ProductionTenantID,
|
||
ProductionKnowledgeBaseID: incoming.ProductionKnowledgeBaseID,
|
||
UpdatedBy: actorFor(r),
|
||
Now: config.Now(),
|
||
}, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrInvalidUISettings) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid UI settings candidate"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not create UI settings candidate"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusCreated, uiSettingsWriteResponse(settings, operation, verification))
|
||
}
|
||
|
||
func activateUISettings(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if r.Header.Get("X-Governance-Action") != "ui-settings" {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "UI settings write intent is required"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var request struct{}
|
||
if err := decodeJSONStrict(r, &request); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid UI settings activation"})
|
||
return
|
||
}
|
||
// Activation applies the revision: if it would flip the endpoints or the
|
||
// production IDs, the caller must be an admin even when they did not author it.
|
||
revision, err := config.Store.GetUISettingsRevision(r.Context(), r.PathValue("id"))
|
||
if err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid UI settings activation"})
|
||
return
|
||
}
|
||
current := operatorServiceFields{}
|
||
if active, err := config.Store.GetActiveUISettings(r.Context()); err == nil {
|
||
current = operatorServiceFieldsOf(active)
|
||
}
|
||
staged := operatorServiceFieldsOf(revision)
|
||
if operatorChangeNeedsAdmin(w, r, current, staged) {
|
||
return
|
||
}
|
||
// Re-verify right before the values go live: a candidate can sit for days and
|
||
// its knowledge base can be deleted in between. An unchanged activation (a
|
||
// replay of an applied revision) skips the check so a retry stays idempotent.
|
||
verification := governance.ServiceTargetVerification{}
|
||
if staged != current {
|
||
var rejected bool
|
||
if verification, rejected = verifyServiceTarget(w, r, config, staged); rejected {
|
||
return
|
||
}
|
||
}
|
||
settings, replayed, err := config.Store.ActivateUISettingsWithOperation(r.Context(), r.PathValue("id"), config.Now(), operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrInvalidUISettings) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid UI settings activation"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not activate UI settings"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, uiSettingsWriteResponse(settings, operation, verification))
|
||
}
|
||
|
||
type createReleaseBatchRequest struct {
|
||
Name string `json:"name"`
|
||
PublicationIDs []string `json:"publication_ids"`
|
||
DeletePublicationIDs []string `json:"delete_publication_ids"`
|
||
ReleaseChangeSetID string `json:"release_change_set_id"`
|
||
// Seal：建批后立即锁定（P0-3，清单已不可变，草稿态只是多按一次）。
|
||
Seal bool `json:"seal"`
|
||
}
|
||
|
||
// reviewBatchStore 是本路由需要的最小能力面：真实 Store 实现它，既有 fake 不必跟着改。
|
||
type reviewBatchStore interface {
|
||
ReviewPublicationsBatchWithOperation(context.Context, governance.ReviewPublicationsBatchRequest, string) (governance.ReviewPublicationsBatchResult, bool, error)
|
||
}
|
||
|
||
// batchReviewCandidates：一次对多条候选做同一个审核决定（批量批准/驳回）。
|
||
// 服务端逐条复用单条审核写入（同一套合法性校验），并逐条回报"哪几条没批成"。
|
||
func batchReviewCandidates(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input struct {
|
||
PublicationIDs []string `json:"publication_ids"`
|
||
Decision string `json:"decision"`
|
||
DecidedVisibility string `json:"decided_visibility"`
|
||
DecisionCode string `json:"decision_code"`
|
||
}
|
||
if err := decodeStrictJSON(r, &input); err != nil || len(input.PublicationIDs) == 0 || len(input.PublicationIDs) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid review batch request"})
|
||
return
|
||
}
|
||
batcher, ok := config.Store.(reviewBatchStore)
|
||
if !ok {
|
||
writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "review batch is not wired"})
|
||
return
|
||
}
|
||
result, replayed, err := batcher.ReviewPublicationsBatchWithOperation(r.Context(), governance.ReviewPublicationsBatchRequest{
|
||
PublicationIDs: input.PublicationIDs, Decision: governance.ReviewState(input.Decision),
|
||
DecidedVisibility: governance.Visibility(input.DecidedVisibility), DecisionCode: input.DecisionCode,
|
||
ReviewerRef: actorFor(r), Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidPublicationReview):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid review batch request"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not record review batch"})
|
||
default:
|
||
operation := "decided"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
}
|
||
|
||
func createReleaseBatch(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input createReleaseBatchRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release batch request"})
|
||
return
|
||
}
|
||
batch, replayed, err := config.Store.CreateReleaseBatchWithOperation(r.Context(), governance.CreateReleaseBatchRequest{
|
||
Name: input.Name, CreatedBy: actorFor(r), PublicationIDs: input.PublicationIDs, DeletePublicationIDs: input.DeletePublicationIDs, ReleaseChangeSetID: input.ReleaseChangeSetID, SealImmediately: input.Seal, Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseBatch):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release batch request"})
|
||
case errors.Is(err, postgresstore.ErrReleaseBatchNotEligible):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "release batch contains ineligible publications"})
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetUnconfirmed):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "release change set must be confirmed and cover every requested publication"})
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetEmpty):
|
||
// 建批时冻结依赖当前实时计划；计划为空就没得冻。
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "current release plan has no releasable item"})
|
||
case errors.Is(err, postgresstore.ErrReleaseBatchGateConflict):
|
||
// 建批即锁定时逐项复核门禁失败：整批已回滚，不会留下半成品。
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "release batch contains ineligible publications"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not create release batch"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusCreated, map[string]any{"data": batch, "operation": operation})
|
||
}
|
||
}
|
||
|
||
type generateReleaseChangeSetRequest struct {
|
||
Name string `json:"name"`
|
||
}
|
||
|
||
func getReleaseAuditChain(w http.ResponseWriter, r *http.Request, store Store) {
|
||
id := r.PathValue("id")
|
||
if !validCatalogID(id) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release execution id"})
|
||
return
|
||
}
|
||
chain, err := store.GetReleaseAuditChain(r.Context(), id)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrReleaseAuditChainNotFound):
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release audit chain not found"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load release audit chain"})
|
||
default:
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": chain})
|
||
}
|
||
}
|
||
|
||
func generateReleaseChangeSet(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input generateReleaseChangeSetRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release change set request"})
|
||
return
|
||
}
|
||
set, replayed, err := config.Store.GenerateReleaseChangeSetWithOperation(r.Context(), governance.GenerateReleaseChangeSetRequest{
|
||
Name: input.Name, CreatedBy: actorFor(r), Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseChangeSet):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release change set request"})
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetEmpty):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "当前发布计划没有可发布项"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not generate release change set"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusCreated, map[string]any{"data": set, "operation": operation})
|
||
}
|
||
}
|
||
|
||
func listReleaseChangeSets(w http.ResponseWriter, r *http.Request, store Store) {
|
||
limit, offset, ok := parseLimitOffset(w, r, 20)
|
||
if !ok {
|
||
return
|
||
}
|
||
page, err := store.ListReleaseChangeSets(r.Context(), governance.ReleaseBatchFilter{Limit: limit, Offset: offset})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list release change sets"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page})
|
||
}
|
||
|
||
func getReleaseChangeSet(w http.ResponseWriter, r *http.Request, store Store, id string) {
|
||
if id == "" {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release change set not found"})
|
||
return
|
||
}
|
||
var detail governance.ReleaseChangeSetDetail
|
||
var err error
|
||
if id == "latest" {
|
||
detail, err = store.LatestReleaseChangeSet(r.Context())
|
||
} else {
|
||
if !validCatalogID(id) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release change set id"})
|
||
return
|
||
}
|
||
detail, err = store.GetReleaseChangeSet(r.Context(), id)
|
||
}
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetNotFound):
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release change set not found"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load release change set"})
|
||
default:
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": detail})
|
||
}
|
||
}
|
||
|
||
func confirmReleaseChangeSet(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input struct{}
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid confirmation request"})
|
||
return
|
||
}
|
||
id := r.PathValue("id")
|
||
set, replayed, err := config.Store.ConfirmReleaseChangeSetWithOperation(r.Context(), governance.ConfirmReleaseChangeSetRequest{
|
||
ChangeSetID: id, ConfirmedBy: actorFor(r), Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseChangeSet):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid confirmation request"})
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetNotFound):
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release change set not found"})
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetStateConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "只有 planned 状态的发布计划可以确认"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not confirm release change set"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": set, "operation": operation})
|
||
}
|
||
}
|
||
|
||
type sealReleaseBatchRequest struct{}
|
||
|
||
func sealReleaseBatch(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input sealReleaseBatchRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid seal request"})
|
||
return
|
||
}
|
||
batch, replayed, err := config.Store.SealReleaseBatchWithOperation(r.Context(), governance.SealReleaseBatchRequest{
|
||
BatchID: r.PathValue("id"), SealedBy: actorFor(r), Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseBatch):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid seal request"})
|
||
case errors.Is(err, postgresstore.ErrReleaseBatchGateConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "release batch no longer satisfies the seal gate"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not seal release batch"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": batch, "operation": operation})
|
||
}
|
||
}
|
||
|
||
type authorizeProductionReleaseRequest struct {
|
||
Confirmation string `json:"confirmation"`
|
||
}
|
||
|
||
type publishNowRequest struct {
|
||
PublicationIDs []string `json:"publication_ids"`
|
||
DeletePublicationIDs []string `json:"delete_publication_ids"`
|
||
Name string `json:"name"`
|
||
Confirmation string `json:"confirmation"`
|
||
}
|
||
|
||
// publishNowPublications：把「建批（冻结 + 锁定）」和「授权生产发布」合成一个动作。
|
||
//
|
||
// 它不放松任何约束——只是把操作员本来要按顺序点两次按钮、再来一次确认语句的流程，
|
||
// 变成一次带确认语句的提交：请求门禁、备份恢复验证门禁、管理员身份、固定确认语句、
|
||
// 发布前快照硬门禁、清单不可变全都照旧由既有两个入口把关。
|
||
func publishNowPublications(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "production release requests are disabled"})
|
||
return
|
||
}
|
||
if !config.ProductionBackupRestoreVerified {
|
||
writeJSON(w, http.StatusPreconditionFailed, map[string]string{"error": "production backup restore verification is required"})
|
||
return
|
||
}
|
||
actor, releaseAuthorized := productionReleaseActor(r, config)
|
||
if !releaseAuthorized {
|
||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "production release authorization required"})
|
||
return
|
||
}
|
||
var input publishNowRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil || input.Confirmation != "publish-to-production" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production release authorization"})
|
||
return
|
||
}
|
||
if len(input.PublicationIDs) == 0 && len(input.DeletePublicationIDs) == 0 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "select at least one publication"})
|
||
return
|
||
}
|
||
seen := make(map[string]bool, len(input.PublicationIDs)+len(input.DeletePublicationIDs))
|
||
for _, publicationID := range append(append([]string{}, input.PublicationIDs...), input.DeletePublicationIDs...) {
|
||
if !validCatalogID(publicationID) || seen[publicationID] || len(seen) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid publication_ids"})
|
||
return
|
||
}
|
||
seen[publicationID] = true
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
name := strings.TrimSpace(input.Name)
|
||
if name == "" || len(name) > 200 || strings.ContainsAny(name, "\x00\r\n") {
|
||
// 固定回退名：同一请求重放时指纹一致，幂等不会被时间戳破坏。
|
||
name = "立即发布"
|
||
}
|
||
now := config.Now().UTC()
|
||
batch, _, err := config.Store.CreateReleaseBatchWithOperation(r.Context(), governance.CreateReleaseBatchRequest{
|
||
Name: name, CreatedBy: actor, PublicationIDs: input.PublicationIDs, DeletePublicationIDs: input.DeletePublicationIDs,
|
||
SealImmediately: true, Now: now,
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseBatch):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release batch request"})
|
||
return
|
||
case errors.Is(err, postgresstore.ErrReleaseBatchNotEligible), errors.Is(err, postgresstore.ErrReleaseBatchGateConflict):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "release batch contains ineligible publications"})
|
||
return
|
||
case errors.Is(err, postgresstore.ErrReleaseChangeSetUnconfirmed), errors.Is(err, postgresstore.ErrReleaseChangeSetEmpty):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "release change set must be confirmed and cover every requested publication"})
|
||
return
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not create release batch"})
|
||
return
|
||
}
|
||
execution, _, err := config.Store.AuthorizeReleaseBatchWithOperation(r.Context(), governance.AuthorizeReleaseBatchRequest{
|
||
BatchID: batch.ID, AuthorizedBy: actor, AuthorizationCode: "production_release_authorized", Now: now,
|
||
}, operationID+":authorize")
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseAuthorization):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production release authorization"})
|
||
case errors.Is(err, postgresstore.ErrReleaseAuthorizationGateConflict):
|
||
// 批次已锁定但不满足生产门禁（例如快照缺失）：如实说明，批次留着不浪费。
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "sealed batch no longer satisfies the production release gate"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not authorize production release"})
|
||
default:
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": map[string]any{"batch": batch, "execution": execution}})
|
||
}
|
||
}
|
||
|
||
func authorizeProductionRelease(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "production release requests are disabled"})
|
||
return
|
||
}
|
||
if !config.ProductionBackupRestoreVerified {
|
||
writeJSON(w, http.StatusPreconditionFailed, map[string]string{"error": "production backup restore verification is required"})
|
||
return
|
||
}
|
||
actor, releaseAuthorized := productionReleaseActor(r, config)
|
||
if !releaseAuthorized {
|
||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "production release authorization required"})
|
||
return
|
||
}
|
||
var input authorizeProductionReleaseRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil || input.Confirmation != "publish-to-production" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production release authorization"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
execution, replayed, err := config.Store.AuthorizeReleaseBatchWithOperation(r.Context(), governance.AuthorizeReleaseBatchRequest{
|
||
BatchID: r.PathValue("id"), AuthorizedBy: actor,
|
||
AuthorizationCode: "production_release_authorized", Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseAuthorization):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production release authorization"})
|
||
case errors.Is(err, postgresstore.ErrReleaseAuthorizationGateConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "sealed batch no longer satisfies the production release gate"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not authorize production release"})
|
||
default:
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": map[string]any{
|
||
"id": execution.ID, "release_batch_id": execution.ReleaseBatchID,
|
||
"operation": execution.Operation, "state": execution.State,
|
||
"target_environment": execution.TargetEnvironment, "job_id": execution.JobID,
|
||
"item_count": execution.ItemCount, "created_at": execution.CreatedAt,
|
||
}, "operation": map[bool]string{true: "replayed", false: "created"}[replayed]})
|
||
}
|
||
}
|
||
|
||
type authorizeProductionRollbackRequest struct {
|
||
Confirmation string `json:"confirmation"`
|
||
}
|
||
|
||
func authorizeProductionRollback(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "production release requests are disabled"})
|
||
return
|
||
}
|
||
actor, releaseAuthorized := productionReleaseActor(r, config)
|
||
if !releaseAuthorized {
|
||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "production release authorization required"})
|
||
return
|
||
}
|
||
var input authorizeProductionRollbackRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil || input.Confirmation != "rollback-production-release" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production rollback authorization"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
execution, replayed, err := config.Store.AuthorizeReleaseRollbackWithOperation(r.Context(), governance.AuthorizeReleaseRollbackRequest{
|
||
PriorExecutionID: r.PathValue("id"), AuthorizedBy: actor,
|
||
AuthorizationCode: "production_rollback_authorized", Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrInvalidReleaseAuthorization):
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production rollback authorization"})
|
||
case errors.Is(err, postgresstore.ErrReleaseAuthorizationGateConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "release execution is not eligible for rollback"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not authorize production rollback"})
|
||
default:
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": map[string]any{
|
||
"id": execution.ID, "release_batch_id": execution.ReleaseBatchID,
|
||
"operation": execution.Operation, "state": execution.State,
|
||
"target_environment": execution.TargetEnvironment, "job_id": execution.JobID,
|
||
"item_count": execution.ItemCount, "created_at": execution.CreatedAt,
|
||
}, "operation": map[bool]string{true: "replayed", false: "created"}[replayed]})
|
||
}
|
||
}
|
||
|
||
type retryProductionReleaseRequest struct {
|
||
Confirmation string `json:"confirmation"`
|
||
}
|
||
|
||
func retryProductionRelease(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
writeJSON(w, http.StatusForbidden, map[string]string{"error": "production release requests are disabled"})
|
||
return
|
||
}
|
||
// 重排沿用原执行的授权人记录（改写它会篡改历史），因此这里只取授权结果。
|
||
if _, releaseAuthorized := productionReleaseActor(r, config); !releaseAuthorized {
|
||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "production release authorization required"})
|
||
return
|
||
}
|
||
var input retryProductionReleaseRequest
|
||
if err := decodeStrictJSON(r, &input); err != nil || input.Confirmation != "retry-production-release" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid production release retry confirmation"})
|
||
return
|
||
}
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
execution, replayed, err := config.Store.RetryReleaseExecutionWithOperation(r.Context(), governance.RetryReleaseExecutionRequest{ExecutionID: r.PathValue("id"), Now: config.Now().UTC()}, operationID)
|
||
if err != nil {
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if errors.Is(err, postgresstore.ErrReleaseExecutionRetryConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "release execution cannot be retried"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "retry release execution"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": safeReleaseExecution(execution), "operation": operation})
|
||
}
|
||
|
||
// productionAuthorizer names the human behind a production double-gate call.
|
||
// With auth enabled the session identity (an admin, enforced by the RBAC
|
||
// matrix) is authoritative; the env-declared ref is only the fallback for the
|
||
// auth-disabled development socket.
|
||
func productionAuthorizer(r *http.Request, config Config) string {
|
||
if identity, ok := sessionFrom(r.Context()); ok {
|
||
return identity.Username
|
||
}
|
||
return config.ProductionReleaseAuthorizerRef
|
||
}
|
||
|
||
// productionReleaseActor authorises a production action and names who performed
|
||
// it. Two sanctioned paths behind the same gates:
|
||
// - the deployment Bearer token (ops scripts) — recorded under the configured
|
||
// fixed authorizer ref;
|
||
// - an authenticated ADMIN session plus the fixed confirmation phrase (the
|
||
// page's 授权发布 / 重排 / 回滚 buttons) — recorded under that operator.
|
||
//
|
||
// The request gate, the backup-restore gate and the worker's execution gate stay
|
||
// independent of this choice: relaxing "who may call" must not relax "may it run".
|
||
func productionReleaseActor(r *http.Request, config Config) (string, bool) {
|
||
if productionReleaseAuthorized(r, config.ProductionReleaseAuthorizationToken) {
|
||
return config.ProductionReleaseAuthorizerRef, true
|
||
}
|
||
if identity, ok := sessionFrom(r.Context()); ok && identity.Role == roleAdmin {
|
||
return identity.Username, true
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
func productionReleaseAuthorized(r *http.Request, token string) bool {
|
||
provided := r.Header.Get("Authorization")
|
||
expected := "Bearer " + token
|
||
return subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) == 1
|
||
}
|
||
|
||
func listReleaseExecutions(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.ReleaseExecutionFilter{Operation: r.URL.Query().Get("operation"), State: r.URL.Query().Get("state"), Limit: 20}
|
||
if filter.Operation != "" && !governance.ReleaseOperation(filter.Operation).Valid() {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "operation must be publish or rollback"})
|
||
return
|
||
}
|
||
if filter.State != "" && !governance.ReleaseExecutionState(filter.State).Valid() {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid release execution state"})
|
||
return
|
||
}
|
||
limit, offset, ok := parseLimitOffset(w, r, 20)
|
||
if !ok {
|
||
return
|
||
}
|
||
filter.Limit, filter.Offset = limit, offset
|
||
page, err := store.ListReleaseExecutions(r.Context(), filter)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list release executions"})
|
||
return
|
||
}
|
||
items := make([]map[string]any, 0, len(page.Items))
|
||
for _, execution := range page.Items {
|
||
items = append(items, safeReleaseExecution(execution))
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": items, "total": page.Total, "limit": limit, "offset": offset})
|
||
}
|
||
|
||
func getReleaseExecution(w http.ResponseWriter, r *http.Request, store Store) {
|
||
detail, err := store.GetReleaseExecution(r.Context(), r.PathValue("id"))
|
||
if errors.Is(err, postgresstore.ErrReleaseExecutionNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release execution not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load release execution"})
|
||
return
|
||
}
|
||
result := safeReleaseExecution(detail.ReleaseExecution)
|
||
items := make([]map[string]any, 0, len(detail.Items))
|
||
for _, item := range detail.Items {
|
||
items = append(items, map[string]any{
|
||
"publication_id": item.PublicationID, "title": item.Title, "action": item.Action, "state": item.State,
|
||
"previous_production_revision": item.PreviousProductionRevision,
|
||
"production_revision": item.ProductionRevision, "error_code": item.ErrorCode,
|
||
"created_at": item.CreatedAt, "completed_at": item.CompletedAt,
|
||
"target_knowledge_base_id": item.TargetKnowledgeBaseID,
|
||
})
|
||
}
|
||
result["items"] = items
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result})
|
||
}
|
||
|
||
func safeReleaseExecution(execution governance.ReleaseExecution) map[string]any {
|
||
return map[string]any{
|
||
"id": execution.ID, "release_batch_id": execution.ReleaseBatchID,
|
||
"operation": execution.Operation, "prior_execution_id": execution.PriorExecutionID,
|
||
"state": execution.State, "target_environment": execution.TargetEnvironment,
|
||
"authorized_by": execution.AuthorizedBy, "job_id": execution.JobID,
|
||
"item_count": execution.ItemCount, "batch_size": execution.BatchSize,
|
||
"current_batch": execution.CurrentBatch, "total_batches": execution.TotalBatches,
|
||
"completed_item_count": execution.CompletedItemCount,
|
||
"failed_item_count": execution.FailedItemCount, "error_code": execution.ErrorCode,
|
||
"failure_level": execution.FailureLevel, "restore_trigger": execution.RestoreTrigger,
|
||
"release_change_set_id": execution.ReleaseChangeSetID,
|
||
"created_at": execution.CreatedAt, "started_at": execution.StartedAt, "completed_at": execution.CompletedAt,
|
||
}
|
||
}
|
||
|
||
func listReleaseBatches(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.ReleaseBatchFilter{State: r.URL.Query().Get("state"), Limit: 20}
|
||
if filter.State != "" && filter.State != "draft" && filter.State != "sealed" && filter.State != "cancelled" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "state must be draft, sealed, or cancelled"})
|
||
return
|
||
}
|
||
limit, offset, ok := parseLimitOffset(w, r, 20)
|
||
if !ok {
|
||
return
|
||
}
|
||
filter.Limit, filter.Offset = limit, offset
|
||
page, err := store.ListReleaseBatches(r.Context(), filter)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list release batches"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page.Items, "total": page.Total, "limit": limit, "offset": offset})
|
||
}
|
||
|
||
func getReleaseBatch(w http.ResponseWriter, r *http.Request, store Store) {
|
||
detail, err := store.GetReleaseBatch(r.Context(), r.PathValue("id"))
|
||
if errors.Is(err, postgresstore.ErrReleaseBatchNotFound) {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "release batch not found"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not load release batch"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": detail})
|
||
}
|
||
|
||
func listTestPublications(w http.ResponseWriter, r *http.Request, store Store) {
|
||
filter := governance.TestPublicationFilter{ProcessingState: r.URL.Query().Get("processing_state"), TargetScope: r.URL.Query().Get("target_scope"), Limit: 50}
|
||
if raw := r.URL.Query().Get("removal"); raw != "" {
|
||
if raw != "required" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "removal must be required"})
|
||
return
|
||
}
|
||
filter.RemovalRequired = true
|
||
}
|
||
if filter.ProcessingState != "" && !validProcessingStateFilter(filter.ProcessingState) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "processing_state must be a comma-separated list of planned, pending, processing, finalizing, completed, failed, stalled, or cancelled"})
|
||
return
|
||
}
|
||
if filter.TargetScope != "" && filter.TargetScope != "internal" && filter.TargetScope != "external" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "target_scope must be internal or external"})
|
||
return
|
||
}
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return
|
||
}
|
||
filter.Limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return
|
||
}
|
||
filter.Offset = value
|
||
}
|
||
page, err := store.ListTestPublications(r.Context(), filter)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list test publications"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{
|
||
"data": page.Items, "total": page.Total, "limit": filter.Limit, "offset": filter.Offset,
|
||
})
|
||
}
|
||
|
||
type confirmChangeSetInput struct {
|
||
ItemSeqs []int64 `json:"item_seqs"`
|
||
Reason string `json:"reason"`
|
||
}
|
||
|
||
func confirmCatalogChangeSet(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input confirmChangeSetInput
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<16))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(&input); err != nil || len(input.ItemSeqs) == 0 || len(input.ItemSeqs) > 200 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid change set confirmation"})
|
||
return
|
||
}
|
||
internalTarget, externalTarget := config.effectiveRoutingTargets(r.Context())
|
||
result, replayed, err := config.Store.ConfirmCatalogChangeSetWithOperation(r.Context(), governance.ConfirmCatalogChangeSetRequest{
|
||
ChangeSetID: r.PathValue("id"), ItemSeqs: input.ItemSeqs,
|
||
OperatorRef: actorFor(r), Reason: input.Reason,
|
||
InternalTargetKBID: internalTarget, ExternalTargetKBID: externalTarget,
|
||
Now: config.Now().UTC(),
|
||
}, operationID)
|
||
if err != nil {
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrCatalogChangeSetNotFound):
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "change set not found"})
|
||
default:
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "could not confirm change set"})
|
||
}
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
|
||
type startTestPublicationsInput struct {
|
||
PublicationIDs []string `json:"publication_ids"`
|
||
Reason string `json:"reason"`
|
||
}
|
||
|
||
func startTestPublications(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input startTestPublicationsInput
|
||
if err := decodeStrictJSON(r, &input); err != nil || len(input.PublicationIDs) == 0 || len(input.PublicationIDs) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid processing confirmation"})
|
||
return
|
||
}
|
||
reason := strings.TrimSpace(input.Reason)
|
||
if reason == "" || len(reason) > 500 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "reason is required"})
|
||
return
|
||
}
|
||
seen := make(map[string]bool, len(input.PublicationIDs))
|
||
for _, publicationID := range input.PublicationIDs {
|
||
if !validCatalogID(publicationID) || seen[publicationID] {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid publication_ids"})
|
||
return
|
||
}
|
||
seen[publicationID] = true
|
||
}
|
||
result, replayed, err := config.Store.StartTestPublicationsWithOperation(r.Context(), governance.StartTestPublicationsRequest{
|
||
PublicationIDs: input.PublicationIDs,
|
||
OperatorRef: actorFor(r),
|
||
Reason: reason,
|
||
Now: config.Now().UTC(),
|
||
}, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not start test processing"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
|
||
func listAcceptance(w http.ResponseWriter, r *http.Request, store Store) {
|
||
records, err := store.ListAcceptanceResults(r.Context(), r.PathValue("id"))
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not list acceptance results"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": records})
|
||
}
|
||
|
||
func listJobFailures(w http.ResponseWriter, r *http.Request, store Store) {
|
||
limit, offset, ok := parseLimitOffset(w, r, 50)
|
||
if !ok {
|
||
return
|
||
}
|
||
page, err := store.ListJobFailures(r.Context(), governance.JobFailureFilter{
|
||
State: r.URL.Query().Get("state"), Limit: limit, Offset: offset,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid failure filter"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": page})
|
||
}
|
||
|
||
// requeueJobFailure re-enqueues one terminal 源头取件 (artifact.fetch) dead
|
||
// letter. 上传解析 dead letters keep their publication-keyed retry path; a
|
||
// fetch failure carries no publication id, so without this route the file is
|
||
// stuck forever after a single terminal fetch failure (expired source session).
|
||
func requeueJobFailure(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
failureID := strings.TrimSpace(r.PathValue("id"))
|
||
if failureID == "" || len(failureID) > 64 || !validCatalogID(failureID) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid failure id"})
|
||
return
|
||
}
|
||
var input struct {
|
||
Reason string `json:"reason"`
|
||
}
|
||
if err := decodeStrictJSON(r, &input); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid requeue request"})
|
||
return
|
||
}
|
||
reason := strings.TrimSpace(input.Reason)
|
||
if reason == "" || len(reason) > 500 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "reason is required"})
|
||
return
|
||
}
|
||
result, replayed, err := config.Store.RequeueArtifactFetchWithOperation(r.Context(), governance.RequeueArtifactFetchRequest{
|
||
FailureID: failureID,
|
||
OperatorRef: actorFor(r),
|
||
Reason: reason,
|
||
Now: config.Now().UTC(),
|
||
}, operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrJobFailureNotRequeueable):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "该失败记录不是可重排的源头取件任务，或已经处理过"})
|
||
return
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not requeue artifact fetch"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result, "replayed": replayed})
|
||
}
|
||
|
||
func retryPublicationsBatch(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input startTestPublicationsInput
|
||
if err := decodeStrictJSON(r, &input); err != nil || len(input.PublicationIDs) == 0 || len(input.PublicationIDs) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid retry batch"})
|
||
return
|
||
}
|
||
reason := strings.TrimSpace(input.Reason)
|
||
if reason == "" || len(reason) > 500 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "reason is required"})
|
||
return
|
||
}
|
||
seen := make(map[string]bool, len(input.PublicationIDs))
|
||
for _, publicationID := range input.PublicationIDs {
|
||
if !validCatalogID(publicationID) || seen[publicationID] {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid publication_ids"})
|
||
return
|
||
}
|
||
seen[publicationID] = true
|
||
}
|
||
result, replayed, err := config.Store.RetryPublicationsBatchWithOperation(r.Context(), governance.RetryPublicationsBatchRequest{
|
||
PublicationIDs: input.PublicationIDs,
|
||
OperatorRef: actorFor(r),
|
||
Reason: reason,
|
||
Now: config.Now().UTC(),
|
||
}, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not retry publications"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
|
||
func cancelPublicationsBatch(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var input startTestPublicationsInput
|
||
if err := decodeStrictJSON(r, &input); err != nil || len(input.PublicationIDs) == 0 || len(input.PublicationIDs) > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid cancel batch"})
|
||
return
|
||
}
|
||
reason := strings.TrimSpace(input.Reason)
|
||
if reason == "" || len(reason) > 500 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "reason is required"})
|
||
return
|
||
}
|
||
seen := make(map[string]bool, len(input.PublicationIDs))
|
||
for _, publicationID := range input.PublicationIDs {
|
||
if !validCatalogID(publicationID) || seen[publicationID] {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid publication_ids"})
|
||
return
|
||
}
|
||
seen[publicationID] = true
|
||
}
|
||
result, replayed, err := config.Store.CancelPublicationsBatchWithOperation(r.Context(), governance.CancelPublicationsBatchRequest{
|
||
PublicationIDs: input.PublicationIDs,
|
||
OperatorRef: actorFor(r),
|
||
Reason: reason,
|
||
Now: config.Now().UTC(),
|
||
}, operationID)
|
||
if errors.Is(err, postgresstore.ErrOperationRequestConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not cancel publications"})
|
||
return
|
||
}
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": result, "operation": operation})
|
||
}
|
||
|
||
func retryPublication(w http.ResponseWriter, r *http.Request, config Config) {
|
||
operationID, ok := requireOperationID(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
publicationID := r.PathValue("id")
|
||
replayed, err := config.Store.RetryPublicationWithOperation(r.Context(), publicationID, config.Now().UTC(), operationID)
|
||
switch {
|
||
case errors.Is(err, postgresstore.ErrOperationRequestConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "Idempotency-Key conflicts with a different request"})
|
||
case errors.Is(err, postgresstore.ErrPublicationNotFound):
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "publication not found"})
|
||
case errors.Is(err, postgresstore.ErrPublicationStateConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]string{"error": "publication is not in a retryable state"})
|
||
case errors.Is(err, postgresstore.ErrPublicationNotRetryable):
|
||
writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "publication has no verified spool artifact"})
|
||
case err != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not retry publication"})
|
||
default:
|
||
operation := "created"
|
||
if replayed {
|
||
operation = "replayed"
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"data": map[string]string{"id": publicationID, "status": "retry_queued"}, "operation": operation})
|
||
}
|
||
}
|
||
|
||
func parseLimitOffset(w http.ResponseWriter, r *http.Request, defaultLimit int) (int, int, bool) {
|
||
limit, offset := defaultLimit, 0
|
||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 1 || value > 100 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be between 1 and 100"})
|
||
return 0, 0, false
|
||
}
|
||
limit = value
|
||
}
|
||
if raw := r.URL.Query().Get("offset"); raw != "" {
|
||
value, err := strconv.Atoi(raw)
|
||
if err != nil || value < 0 || value > 1000000 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "offset must be between 0 and 1000000"})
|
||
return 0, 0, false
|
||
}
|
||
offset = value
|
||
}
|
||
return limit, offset, true
|
||
}
|
||
|
||
func decodeStrictJSON(r *http.Request, destination any) error {
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 64<<10))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(destination); err != nil {
|
||
return err
|
||
}
|
||
var trailing any
|
||
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
|
||
return errors.New("trailing JSON value")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func validProcessingState(state string) bool {
|
||
switch state {
|
||
case "planned", "pending", "processing", "finalizing", "completed", "failed", "stalled", "cancelled":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// validProcessingStateFilter accepts a single state or a comma-separated list so
|
||
// callers can ask for several lanes at once (e.g. the attention lane needs
|
||
// failed+stalled in one query instead of paging the whole queue).
|
||
func validProcessingStateFilter(filter string) bool {
|
||
for _, state := range strings.Split(filter, ",") {
|
||
if !validProcessingState(state) {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func newID() (string, error) {
|
||
var raw [16]byte
|
||
if _, err := rand.Read(raw[:]); err != nil {
|
||
return "", fmt.Errorf("read random ID: %w", err)
|
||
}
|
||
raw[6] = (raw[6] & 0x0f) | 0x40
|
||
raw[8] = (raw[8] & 0x3f) | 0x80
|
||
encoded := hex.EncodeToString(raw[:])
|
||
return encoded[0:8] + "-" + encoded[8:12] + "-" + encoded[12:16] + "-" + encoded[16:20] + "-" + encoded[20:32], nil
|
||
}
|
||
|
||
func decodeJSONStrict(r *http.Request, target any) error {
|
||
decoder := json.NewDecoder(io.LimitReader(r.Body, 16*1024))
|
||
decoder.DisallowUnknownFields()
|
||
if err := decoder.Decode(target); err != nil {
|
||
return err
|
||
}
|
||
var trailing any
|
||
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
|
||
if err == nil {
|
||
return errors.New("multiple JSON values are not allowed")
|
||
}
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(status)
|
||
_ = json.NewEncoder(w).Encode(value)
|
||
}
|