姚章浩
3f603aa293
需求：整库逻辑备份做定时，但要在「系统设置 → 存储与备份」里设置定时时间与储存天数。 后端 - migration 000045 backup_schedule（单行：enabled/schedule_time/retention_days/retention_months/ last_applied_*），store + 领域校验（HH:MM、1~365 天、0~24 月，非法值 400 且不落库不下发） - internal/backuphost：通过 ssh 在目标主机上读回/下发计划并立即备份（未接线时 Available=false， 页面降级提示；主机侧错误带输出回传）。最近一次只认有 MANIFEST.json 的完整目录，跳 *.incomplete - 接口 GET/PUT /api/v1/backup-schedule、GET …/host、POST …/apply、POST …/run； 写操作经 classifyRole 限管理员；run 为异步 202 + 在途单飞（http 写超时 30s，备份约 35s） - openapi.yaml 补齐 5 条路由（契约测试要求每条都写 400/500） 主机侧 - scripts/install_backup_schedule.sh（幂等）：写 /etc/knowledge-governance/backup-schedule.env 与 /etc/cron.d/kb-logical-backup，参数校验，--show/--disable，最后一行 SCHEDULE_APPLIED 供平台核对 - logical-backup.sh 从该文件读保留策略；失败或被中断把目录改名 *.incomplete，不留'像成功'的半成品 - install --show 只回显计划行（原先把 SHELL=/PATH= 当计划显示） 前端 - web/src/backupSchedule.ts（严格解析：缺计划字段即报错，不把默认值伪装成设置） - BackupScheduleSection：容器 + 纯视图（管理员表单/只读、已一致/待下发、在途轮询） 运行前提（已落实）：运行镜像加 openssh-client，且容器 uid 1000 在 /etc/passwd 有名字 （否则 ssh 报 No user exists for uid 1000）；凭据只读挂载 secrets/backup-host（git 忽略） 实测：主机 cron 每天 03:30 / 保留 7 天 / 每月 3 份；立即备份一次 34 秒 / 742.7 MiB / 失败 0； Go 与前端测试全绿（170 vitest）。
2366 lines
103 KiB
Go
2366 lines
103 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 {
|
|
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 负责在目标库
|
|
// 所在主机上读回/下发计划并触发一次整库备份（未接线时页面降级展示）。
|
|
BackupSchedule BackupScheduleStore
|
|
BackupHost BackupHost
|
|
BackupRuns *BackupRunTracker
|
|
ProductionBindings ProductionBindingStore
|
|
ProductionKnowledgeBases ProductionKnowledgeBaseLister
|
|
ProductionWeKnoraBaseURL string
|
|
ServiceTargetVerifier ServiceTargetVerifier
|
|
Now func() time.Time
|
|
RequireWeKnoraReady bool
|
|
ProductionReleaseRequestsEnabled bool
|
|
ProductionBackupRestoreVerified bool
|
|
ProductionReleaseAuthorizerRef string
|
|
ProductionReleaseAuthorizationToken string
|
|
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 /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/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-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("GET /api/v1/review-candidates", func(w http.ResponseWriter, r *http.Request) {
|
|
listReviewCandidates(w, r, config.Store)
|
|
})
|
|
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/{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]bool{
|
|
"production_release_requests_enabled": config.ProductionReleaseRequestsEnabled,
|
|
"production_backup_restore_verified": config.ProductionBackupRestoreVerified,
|
|
}})
|
|
})
|
|
mux.HandleFunc("GET /api/v1/test-publications", func(w http.ResponseWriter, r *http.Request) {
|
|
listTestPublications(w, r, config.Store)
|
|
})
|
|
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
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"data": sources, "summary": map[string]int{"total": len(sources)}})
|
|
}
|
|
|
|
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"`
|
|
}
|
|
|
|
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, 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 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"`
|
|
}
|
|
|
|
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 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)
|
|
}
|