Adopt ConfigKit for pipeline loading
This commit is contained in:
391
internal/config/pipeline_configkit.go
Normal file
391
internal/config/pipeline_configkit.go
Normal file
@@ -0,0 +1,391 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
configkit "gitea.maximumdirect.net/eric/ConfigKit"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
||||
)
|
||||
|
||||
type pipelineResolutionMetadata struct {
|
||||
rootPath string
|
||||
imports []string
|
||||
sources []string
|
||||
selectedProfile *pipelineProfileSelection
|
||||
effectiveDigest string
|
||||
logicalNotariusConfig string
|
||||
logicalNotariusWorking string
|
||||
logicalNotariusCaptured bool
|
||||
ownership []pipelineFieldOwnership
|
||||
artifactFamilies map[string]ScriptoriumArtifactFamilyConfig
|
||||
artifactFamiliesExpanded bool
|
||||
publishDeclared bool
|
||||
configkitMetadata configkit.Metadata
|
||||
configkitSources []configkit.SourceInfo
|
||||
}
|
||||
|
||||
type pipelineProfileSelection struct {
|
||||
name string
|
||||
source string
|
||||
overlayPath string
|
||||
}
|
||||
|
||||
type pipelineFieldOwnership struct {
|
||||
path string
|
||||
sources []string
|
||||
}
|
||||
|
||||
func loadPipelineWithConfigKit(path string, opts PipelineLoadOptions) (*PipelineConfig, error) {
|
||||
profile := selectedPipelineProfileName(opts)
|
||||
result, err := configkit.Load(configkit.File(path), configkit.Options[PipelineConfig]{
|
||||
Imports: true,
|
||||
Profile: profile,
|
||||
Definition: configkit.Definition[PipelineConfig]{
|
||||
Defaults: pipelineConfigDefaults,
|
||||
Resolve: resolvePipelineConfig,
|
||||
Validate: validatePipelineConfigKit,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, describePipelineLoadError(err)
|
||||
}
|
||||
cfg := result.Config
|
||||
metadata, err := pipelineMetadataFromConfigKit(&cfg, result, opts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load pipeline config: %w", err)
|
||||
}
|
||||
cfg.resolution = metadata
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
func describePipelineLoadError(err error) error {
|
||||
var loadError *configkit.LoadError
|
||||
if !errors.As(err, &loadError) {
|
||||
return fmt.Errorf("load pipeline config: %w", err)
|
||||
}
|
||||
var diagnostics bytes.Buffer
|
||||
resolved := loadError.Diagnostics()
|
||||
sourcePaths := make(map[string]string, len(loadError.Sources()))
|
||||
for _, source := range loadError.Sources() {
|
||||
if source.Path != "" {
|
||||
sourcePaths[source.ID] = source.Path
|
||||
} else if source.Label != "" {
|
||||
sourcePaths[source.ID] = source.Label
|
||||
}
|
||||
}
|
||||
for index := range resolved {
|
||||
if resolved[index].Primary != nil {
|
||||
if name := sourcePaths[resolved[index].Primary.SourceID]; name != "" {
|
||||
resolved[index].Primary.SourceID = name
|
||||
}
|
||||
}
|
||||
for relatedIndex := range resolved[index].Related {
|
||||
if name := sourcePaths[resolved[index].Related[relatedIndex].SourceID]; name != "" {
|
||||
resolved[index].Related[relatedIndex].SourceID = name
|
||||
}
|
||||
}
|
||||
}
|
||||
if writeErr := configkit.WriteDiagnostics(&diagnostics, resolved); writeErr != nil {
|
||||
return fmt.Errorf("load pipeline config: %w", err)
|
||||
}
|
||||
return fmt.Errorf("load pipeline config: strict decode failed or validation failed: %w: %s", err, strings.TrimSpace(diagnostics.String()))
|
||||
}
|
||||
|
||||
func selectedPipelineProfileName(opts PipelineLoadOptions) string {
|
||||
if opts.Profile == nil || *opts.Profile == "" || *opts.Profile == "default" {
|
||||
return ""
|
||||
}
|
||||
return *opts.Profile
|
||||
}
|
||||
|
||||
func pipelineConfigDefaults() PipelineConfig {
|
||||
return PipelineConfig{
|
||||
Workspace: WorkspaceConfig{Root: DefaultWorkspaceRoot},
|
||||
Campaigns: CampaignsConfig{Root: DefaultCampaignsRoot},
|
||||
Storage: StorageConfig{Backend: StorageBackendLocal},
|
||||
Spool: SpoolConfig{Root: DefaultSpoolRoot},
|
||||
Cache: CacheConfig{Root: DefaultCacheRoot, S3Audio: boolPtr(DefaultCacheS3Audio)},
|
||||
Publish: &PublishConfig{
|
||||
Enabled: boolPtr(DefaultArchiveEnabled), UploadRun: boolPtr(DefaultArchiveUploadRun),
|
||||
Outputs: defaultPublishOutputs(),
|
||||
},
|
||||
WhisperX: WhisperXConfig{
|
||||
Language: DefaultWhisperXLanguage, Timeout: DefaultWhisperXTimeout,
|
||||
RetryDelay: DefaultWhisperXRetryDelay, Concurrency: intPtr(DefaultWhisperXConcurrency),
|
||||
Retries: intPtr(DefaultWhisperXRetries),
|
||||
},
|
||||
Seriatim: SeriatimConfig{
|
||||
Binary: DefaultSeriatimBinary, Timeout: DefaultSeriatimTimeout,
|
||||
OutputSchema: DefaultSeriatimOutputSchema, CoalesceGap: float64Ptr(DefaultSeriatimCoalesceGap),
|
||||
Report: boolPtr(DefaultSeriatimReport),
|
||||
},
|
||||
Audita: AuditaConfig{Binary: DefaultAuditaBinary, Timeout: DefaultAuditaTimeout, Report: boolPtr(DefaultAuditaReport)},
|
||||
Normalize: &NormalizeConfig{OutputPath: DefaultNormalizeOutputPath, OutputSchema: DefaultNormalizeOutputSchema, Report: boolPtr(DefaultNormalizeReport)},
|
||||
Trim: &TrimConfig{
|
||||
Enabled: boolPtr(DefaultTrimEnabled), OutputPath: DefaultTrimOutputPath,
|
||||
Bounds: TrimBoundsConfig{
|
||||
PromptID: DefaultTrimBoundsPromptID, TranscriptInputName: DefaultTrimBoundsTranscriptInputName,
|
||||
OutputPath: DefaultTrimBoundsOutputPath, Timeout: DefaultTrimBoundsTimeout,
|
||||
},
|
||||
Seriatim: TrimSeriatimConfig{Report: boolPtr(DefaultTrimSeriatimReport)},
|
||||
},
|
||||
Render: &RenderConfig{
|
||||
Enabled: boolPtr(DefaultRenderEnabled), Format: DefaultRenderFormat, Title: DefaultRenderTitle,
|
||||
IncludeTimestamps: boolPtr(DefaultRenderTimestamps), IncludeSegmentIDs: boolPtr(DefaultRenderSegmentIDs),
|
||||
},
|
||||
Notification: NotificationConfig{Mode: DefaultNotificationMode},
|
||||
}
|
||||
}
|
||||
|
||||
func defaultPublishOutputs() []PublishOutputRule {
|
||||
outputs := append([]PublishOutputRule(nil), DefaultPublishOutputs...)
|
||||
for index := range outputs {
|
||||
outputs[index].Required = boolPtr(true)
|
||||
}
|
||||
return outputs
|
||||
}
|
||||
|
||||
func resolvePipelineConfig(cfg *PipelineConfig, resolution *configkit.Resolution) error {
|
||||
backend := strings.ToLower(strings.TrimSpace(cfg.Storage.Backend))
|
||||
if backend != cfg.Storage.Backend {
|
||||
configkit.Normalize(resolution, configkit.Keys("storage", "backend"), &cfg.Storage.Backend, backend, "canonical storage backend")
|
||||
}
|
||||
if cfg.Storage.S3 != nil {
|
||||
path := configkit.Keys("storage", "s3")
|
||||
configkit.Default(resolution, path.Key("root_prefix"), &cfg.Storage.S3.RootPrefix, DefaultStorageS3RootPrefix, "default S3 root prefix")
|
||||
configkit.Default(resolution, path.Key("access_key_id_env"), &cfg.Storage.S3.AccessKeyIDEnv, DefaultS3AccessKeyIDEnv, "default S3 access-key reference")
|
||||
configkit.Default(resolution, path.Key("secret_access_key_env"), &cfg.Storage.S3.SecretKeyEnv, DefaultS3SecretAccessKeyEnv, "default S3 secret-key reference")
|
||||
}
|
||||
if cfg.Trim != nil && cfg.Trim.Enabled != nil && *cfg.Trim.Enabled && cfg.Scriptorium == nil {
|
||||
value := &ScriptoriumConfig{Binary: DefaultScriptoriumBinary, Timeout: DefaultScriptoriumTimeout}
|
||||
configkit.Default(resolution, configkit.Keys("scriptorium"), &cfg.Scriptorium, value, "trim requires Scriptorium")
|
||||
}
|
||||
if cfg.Scriptorium != nil {
|
||||
path := configkit.Keys("scriptorium")
|
||||
configkit.Default(resolution, path.Key("binary"), &cfg.Scriptorium.Binary, DefaultScriptoriumBinary, "default Scriptorium binary")
|
||||
configkit.Default(resolution, path.Key("timeout"), &cfg.Scriptorium.Timeout, DefaultScriptoriumTimeout, "default Scriptorium timeout")
|
||||
}
|
||||
if cfg.Notarius != nil {
|
||||
path := configkit.Keys("notarius")
|
||||
configkit.Default(resolution, path.Key("binary"), &cfg.Notarius.Binary, DefaultNotariusBinary, "default Notarius binary")
|
||||
configkit.Default(resolution, path.Key("timeout"), &cfg.Notarius.Timeout, DefaultNotariusTimeout, "default Notarius timeout")
|
||||
if cfg.Notarius.Enabled && cfg.Notarius.ConfigPath != "" {
|
||||
configkit.Default(resolution, path.Key("working_directory"), &cfg.Notarius.WorkingDirectory,
|
||||
filepath.Dir(cfg.Notarius.ConfigPath), "derived from Notarius config path", path.Key("config_path"))
|
||||
}
|
||||
}
|
||||
if cfg.Publish != nil {
|
||||
defaultedOutput := false
|
||||
for index := range cfg.Publish.Outputs {
|
||||
output := cfg.Publish.Outputs[index]
|
||||
path := configkit.Keys("publish", "outputs").Index(index).Key("required")
|
||||
if !resolution.IsExplicit(path) {
|
||||
defaultedOutput = true
|
||||
}
|
||||
configkit.Default(resolution, path, &output.Required, boolPtr(true), "publish outputs are required by default")
|
||||
cfg.Publish.Outputs[index] = output
|
||||
}
|
||||
if defaultedOutput {
|
||||
resolution.Record(configkit.Keys("publish", "outputs"), configkit.ChangeDerivedDefault, "publish output entry defaults")
|
||||
}
|
||||
}
|
||||
before := cloneNotariusForResolution(cfg.Notarius)
|
||||
if err := NormalizePipelineConfig(cfg); err != nil {
|
||||
resolution.Error(configkit.Keys("notarius"), err.Error())
|
||||
return nil
|
||||
}
|
||||
if !reflect.DeepEqual(before, cfg.Notarius) {
|
||||
resolution.Record(configkit.Keys("notarius"), configkit.ChangeNormalization, "canonical Notarius configuration")
|
||||
}
|
||||
if cfg.Publish != nil {
|
||||
configuredOutputs := scriptoriumOutputPathMap(cfg.Scriptorium)
|
||||
extractionOutputs := notariusOutputKeySet(cfg.Notarius)
|
||||
for index := range cfg.Publish.Outputs {
|
||||
output := cfg.Publish.Outputs[index]
|
||||
if strings.TrimSpace(output.Dest) != "" {
|
||||
continue
|
||||
}
|
||||
destination, err := artifactpolicy.ResolvePublishedDestinationWithExtractions(strings.TrimSpace(output.Source), "", configuredOutputs, extractionOutputs)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
configkit.Normalize(resolution, configkit.Keys("publish", "outputs").Index(index).Key("dest"), &output.Dest, destination, "derived from publish source")
|
||||
cfg.Publish.Outputs[index] = output
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func cloneNotariusForResolution(value *NotariusConfig) *NotariusConfig {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
clone := *value
|
||||
clone.References = cloneStringMap(value.References)
|
||||
if value.Outputs != nil {
|
||||
clone.Outputs = make(map[string]NotariusOutputConfig, len(value.Outputs))
|
||||
for key, output := range value.Outputs {
|
||||
clone.Outputs[key] = output
|
||||
}
|
||||
}
|
||||
return &clone
|
||||
}
|
||||
|
||||
func cloneStringMap(value map[string]string) map[string]string {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
clone := make(map[string]string, len(value))
|
||||
for key, item := range value {
|
||||
clone[key] = item
|
||||
}
|
||||
return clone
|
||||
}
|
||||
|
||||
func validatePipelineConfigKit(cfg *PipelineConfig, validation *configkit.Validation) {
|
||||
// Checks that depend on family expansion or campaign/session context remain
|
||||
// at Narratio's explicit resolution boundary. This early definition check is
|
||||
// deliberately limited to an invariant that is wholly file-local.
|
||||
if strings.TrimSpace(cfg.Workspace.Root) == "" {
|
||||
validation.Error(configkit.Keys("workspace", "root"), "workspace root is required")
|
||||
}
|
||||
}
|
||||
|
||||
func pipelineMetadataFromConfigKit(cfg *PipelineConfig, result *configkit.Result[PipelineConfig], opts PipelineLoadOptions) (*pipelineResolutionMetadata, error) {
|
||||
sources := result.Sources()
|
||||
metadata := result.Metadata()
|
||||
if err := rejectDuplicateBaseDefinitions(cfg, metadata, sources); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved := &pipelineResolutionMetadata{
|
||||
configkitMetadata: metadata, configkitSources: sources,
|
||||
publishDeclared: metadata.IsExplicit(configkit.Keys("publish")),
|
||||
}
|
||||
pathsByID := make(map[string]string, len(sources))
|
||||
for index, source := range sources {
|
||||
path := source.Path
|
||||
if path == "" {
|
||||
path = source.Label
|
||||
}
|
||||
pathsByID[source.ID] = path
|
||||
resolved.sources = append(resolved.sources, path)
|
||||
if index == 0 {
|
||||
resolved.rootPath = path
|
||||
} else if source.Layer == configkit.LayerBase {
|
||||
resolved.imports = append(resolved.imports, path)
|
||||
}
|
||||
if source.Layer == configkit.LayerProfile {
|
||||
resolved.selectedProfile = &pipelineProfileSelection{name: selectedPipelineProfileName(opts), source: "explicit", overlayPath: path}
|
||||
}
|
||||
}
|
||||
detailed, err := pipelineMetadataPaths(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, value := range detailed {
|
||||
info, ok := metadata.Lookup(configkit.Keys(value.segments...))
|
||||
sourceNames := []string{pipelineDefaultOwnershipSource}
|
||||
if ok {
|
||||
sourceNames = ownershipSources(info.Origin, pathsByID)
|
||||
}
|
||||
resolved.ownership = append(resolved.ownership, pipelineFieldOwnership{path: value.record.Path, sources: sourceNames})
|
||||
}
|
||||
return resolved, nil
|
||||
}
|
||||
|
||||
func pipelineMetadataPaths(cfg *PipelineConfig) ([]projectedPipelineValue, error) {
|
||||
effective, err := projectionDetailedRecords(projectEffectivePipeline(cfg, false))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if cfg.Scriptorium == nil || len(cfg.Scriptorium.ArtifactFamilies) == 0 {
|
||||
return effective, nil
|
||||
}
|
||||
families, err := projectionDetailedRecords(map[string]any{"scriptorium": map[string]any{"artifact_families": cfg.Scriptorium.ArtifactFamilies}})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return append(effective, families...), nil
|
||||
}
|
||||
|
||||
func ownershipSources(origin configkit.Origin, pathsByID map[string]string) []string {
|
||||
if origin.Layer == configkit.LayerDefault || (origin.Layer == configkit.LayerResolution && origin.Operation == configkit.ChangeDerivedDefault) {
|
||||
return []string{pipelineDefaultOwnershipSource}
|
||||
}
|
||||
if origin.Layer == configkit.LayerResolution {
|
||||
return []string{"resolution"}
|
||||
}
|
||||
seen := map[string]struct{}{}
|
||||
var sources []string
|
||||
for _, location := range origin.Locations {
|
||||
value := pathsByID[location.SourceID]
|
||||
if value == "" {
|
||||
value = location.SourceID
|
||||
}
|
||||
if _, ok := seen[value]; !ok {
|
||||
seen[value] = struct{}{}
|
||||
sources = append(sources, value)
|
||||
}
|
||||
}
|
||||
if len(sources) == 0 {
|
||||
return []string{pipelineDefaultOwnershipSource}
|
||||
}
|
||||
return sources
|
||||
}
|
||||
|
||||
func rejectDuplicateBaseDefinitions(cfg *PipelineConfig, metadata configkit.Metadata, sources []configkit.SourceInfo) error {
|
||||
pathsByID := make(map[string]string, len(sources))
|
||||
for _, source := range sources {
|
||||
pathsByID[source.ID] = source.Path
|
||||
}
|
||||
var declarations []struct {
|
||||
kind string
|
||||
key string
|
||||
path configkit.Path
|
||||
}
|
||||
if cfg.Scriptorium != nil {
|
||||
for key := range cfg.Scriptorium.Artifacts {
|
||||
declarations = append(declarations, struct {
|
||||
kind, key string
|
||||
path configkit.Path
|
||||
}{"artifact", key, configkit.Keys("scriptorium", "artifacts", key)})
|
||||
}
|
||||
for key := range cfg.Scriptorium.ArtifactFamilies {
|
||||
declarations = append(declarations, struct {
|
||||
kind, key string
|
||||
path configkit.Path
|
||||
}{"artifact family", key, configkit.Keys("scriptorium", "artifact_families", key)})
|
||||
}
|
||||
}
|
||||
sort.Slice(declarations, func(i, j int) bool { return declarations[i].path.String() < declarations[j].path.String() })
|
||||
for _, declaration := range declarations {
|
||||
seen := map[string]struct{}{}
|
||||
for _, change := range metadata.History(declaration.path) {
|
||||
if change.Origin.Layer != configkit.LayerBase {
|
||||
continue
|
||||
}
|
||||
for _, location := range change.Origin.Locations {
|
||||
seen[location.SourceID] = struct{}{}
|
||||
}
|
||||
}
|
||||
if len(seen) > 1 {
|
||||
claims := make([]string, 0, len(seen))
|
||||
for sourceID := range seen {
|
||||
claims = append(claims, pathsByID[sourceID])
|
||||
}
|
||||
sort.Strings(claims)
|
||||
return fmt.Errorf("pipeline Scriptorium %s %q is declared by multiple base files: %s", declaration.kind, declaration.key, strings.Join(claims, ", "))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func intPtr(value int) *int { return &value }
|
||||
func float64Ptr(value float64) *float64 { return &value }
|
||||
func boolPtr(value bool) *bool { return &value }
|
||||
Reference in New Issue
Block a user