635 lines
24 KiB
Go
635 lines
24 KiB
Go
package checkpoint
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"path"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
|
|
coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
|
)
|
|
|
|
type WorkspaceRecorder struct {
|
|
root string
|
|
identityDigest string
|
|
now func() time.Time
|
|
}
|
|
|
|
func NewWorkspaceRecorder(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointRecorder, error) {
|
|
root, err := settings.CheckpointDirectory(identity)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(root) == "" {
|
|
return pipeline.NoopCheckpointRecorder(), nil
|
|
}
|
|
return &WorkspaceRecorder{root: root, identityDigest: identity.Digest, now: time.Now}, nil
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) SourceRunning(moduleKey string) error {
|
|
manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusRunning)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.StartedAt = timePtr(r.timestamp())
|
|
return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) SourceSucceeded(moduleKey string, doc *source.SourceDocument) error {
|
|
if doc == nil {
|
|
return fmt.Errorf("checkpoint source document must not be nil")
|
|
}
|
|
if err := r.writePayload("source/source-document.json", sourceDocumentEnvelope{Document: cloneSourceDocument(*doc)}); err != nil {
|
|
return err
|
|
}
|
|
manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusSucceeded)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.OutputDigests = workspaceFingerprints(digestFingerprints("source_document", doc.Digest))
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{
|
|
StageManifest: manifest,
|
|
SourceID: doc.ID,
|
|
})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error {
|
|
manifest := r.newStageManifest(coreworkspace.StageSource, coreworkspace.StatusFailed)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
manifest.Metadata = errorMetadata(err)
|
|
return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ChunkRunning(moduleKey string, sourceDigest string) error {
|
|
manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusRunning)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest))
|
|
manifest.StartedAt = timePtr(r.timestamp())
|
|
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string, chunks []contracts.SourceChunk, warnings []contracts.Warning) error {
|
|
payload := chunksEnvelope{Chunks: chunkEnvelopes(chunks), Warnings: cloneWarnings(warnings)}
|
|
if err := r.writePayload("chunk/chunks.json", payload); err != nil {
|
|
return err
|
|
}
|
|
manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest))
|
|
manifest.OutputDigests = workspaceFingerprints(chunkOutputDigests(chunks))
|
|
manifest.ValidationStatus = validationStatusString(warnings, nil)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{
|
|
StageManifest: manifest,
|
|
ChunkCount: len(chunks),
|
|
})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error {
|
|
manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceededWithRejections)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest))
|
|
manifest.ValidationStatus = "rejected"
|
|
manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected})
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ChunkFailed(moduleKey string, sourceDigest string, err error) error {
|
|
manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusFailed)
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest))
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
manifest.Metadata = errorMetadata(err)
|
|
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error {
|
|
manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies)
|
|
manifest.StartedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ExtractSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []contracts.ExtractOutput, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
|
|
payload := extractOutputsEnvelope{
|
|
Outputs: extractOutputEnvelopes(outputs),
|
|
Rejected: cloneRejectedOutputs(rejected),
|
|
Warnings: cloneWarnings(warnings),
|
|
}
|
|
if err := r.writePayload(lanePayloadPath("extract", laneID, "outputs.json"), payload); err != nil {
|
|
return err
|
|
}
|
|
manifest := r.laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies)
|
|
manifest.OutputDigests = workspaceFingerprints(rawOutputDigests(extractPayloads(outputs)))
|
|
manifest.ValidationStatus = validationStatusString(warnings, rejected)
|
|
manifest.Rejections = rejectionSummaries(rejected)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{
|
|
StageManifest: manifest,
|
|
ChunkCount: len(outputs) + len(rejected),
|
|
OutputCount: len(outputs),
|
|
})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error {
|
|
manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusFailed, laneID, moduleKey, dependencies)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
manifest.Metadata = errorMetadata(err)
|
|
return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) MergeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error {
|
|
manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusRunning, laneID, moduleKey, dependencies)
|
|
manifest.StartedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) MergeSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output contracts.MergeOutput, warnings []contracts.Warning) error {
|
|
payload := mergeOutputEnvelope{Output: mergeOutputEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}
|
|
if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), payload); err != nil {
|
|
return err
|
|
}
|
|
manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies)
|
|
manifest.OutputDigests = workspaceFingerprints(rawOutputDigests([]contracts.RawPayload{output.Payload}))
|
|
manifest.ValidationStatus = validationStatusString(warnings, nil)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{
|
|
StageManifest: manifest,
|
|
InputCount: len(dependencies),
|
|
})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error {
|
|
manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies)
|
|
manifest.ValidationStatus = "rejected"
|
|
manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected})
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error {
|
|
manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusFailed, laneID, moduleKey, dependencies)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
manifest.Metadata = errorMetadata(err)
|
|
return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) NormalizeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error {
|
|
manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusRunning, laneID, moduleKey, dependencies)
|
|
manifest.StartedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) NormalizeSucceeded(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output contracts.NormalizeOutput, warnings []contracts.Warning) error {
|
|
payload := normalizeOutputEnvelope{Output: normalizeOutputEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}
|
|
if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), payload); err != nil {
|
|
return err
|
|
}
|
|
manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies)
|
|
manifest.OutputDigests = workspaceFingerprints(rawOutputDigests([]contracts.RawPayload{output.Payload}))
|
|
manifest.ValidationStatus = validationStatusString(warnings, nil)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error {
|
|
manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies)
|
|
manifest.ValidationStatus = "rejected"
|
|
manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected})
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error {
|
|
manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusFailed, laneID, moduleKey, dependencies)
|
|
manifest.CompletedAt = timePtr(r.timestamp())
|
|
manifest.Metadata = errorMetadata(err)
|
|
return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest})
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) writeManifest(name string, payload any) error {
|
|
return r.writeJSON(name, payload)
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) writePayload(name string, payload any) error {
|
|
return r.writeJSON(name, payload)
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) writeJSON(name string, payload any) error {
|
|
if r == nil || strings.TrimSpace(r.root) == "" {
|
|
return nil
|
|
}
|
|
return coreworkspace.WriteJSON(r.root, name, payload)
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) timestamp() time.Time {
|
|
if r == nil || r.now == nil {
|
|
return time.Now().UTC()
|
|
}
|
|
return r.now().UTC()
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) newStageManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus) coreworkspace.StageManifest {
|
|
manifest := coreworkspace.NewStageManifest(stage, status)
|
|
if strings.TrimSpace(r.identityDigest) != "" {
|
|
manifest.Metadata = map[string]string{"checkpoint_identity_digest": r.identityDigest}
|
|
}
|
|
return manifest
|
|
}
|
|
|
|
func (r *WorkspaceRecorder) laneManifest(stage coreworkspace.StageName, status coreworkspace.StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) coreworkspace.StageManifest {
|
|
manifest := r.newStageManifest(stage, status)
|
|
manifest.LaneID = laneID
|
|
manifest.ModuleKey = moduleKey
|
|
manifest.DependencyFingerprints = workspaceFingerprints(dependencies)
|
|
return manifest
|
|
}
|
|
|
|
type sourceDocumentEnvelope struct {
|
|
Document source.SourceDocument `json:"document"`
|
|
}
|
|
|
|
type chunksEnvelope struct {
|
|
Chunks []chunkEnvelope `json:"chunks"`
|
|
Warnings []contracts.Warning `json:"warnings,omitempty"`
|
|
}
|
|
|
|
type chunkEnvelope struct {
|
|
ID string `json:"id"`
|
|
SourceID string `json:"source_id"`
|
|
Index int `json:"index"`
|
|
StartUnitID int `json:"start_unit_id"`
|
|
EndUnitID int `json:"end_unit_id"`
|
|
Content binaryEnvelope `json:"content"`
|
|
Units []source.SourceUnit `json:"units,omitempty"`
|
|
Metadata map[string]any `json:"metadata,omitempty"`
|
|
}
|
|
|
|
type extractOutputsEnvelope struct {
|
|
Outputs []extractOutputEnvelope `json:"outputs"`
|
|
Rejected []contracts.RejectedOutput `json:"rejected,omitempty"`
|
|
Warnings []contracts.Warning `json:"warnings,omitempty"`
|
|
}
|
|
|
|
type extractOutputEnvelope struct {
|
|
LaneID string `json:"lane_id"`
|
|
ExtractorKey string `json:"extractor_key"`
|
|
SourceID string `json:"source_id"`
|
|
ChunkID string `json:"chunk_id"`
|
|
ChunkIndex int `json:"chunk_index"`
|
|
Schema contracts.ResponseSchema `json:"schema,omitempty"`
|
|
Payload binaryEnvelope `json:"payload"`
|
|
}
|
|
|
|
type mergeOutputEnvelope struct {
|
|
Output mergeOutputPayload `json:"output"`
|
|
Warnings []contracts.Warning `json:"warnings,omitempty"`
|
|
}
|
|
|
|
type mergeOutputPayload struct {
|
|
LaneID string `json:"lane_id"`
|
|
MergerKey string `json:"merger_key"`
|
|
SourceID string `json:"source_id,omitempty"`
|
|
Schema contracts.ResponseSchema `json:"schema,omitempty"`
|
|
Payload binaryEnvelope `json:"payload"`
|
|
}
|
|
|
|
type normalizeOutputEnvelope struct {
|
|
Output normalizeOutputPayload `json:"output"`
|
|
Warnings []contracts.Warning `json:"warnings,omitempty"`
|
|
}
|
|
|
|
type normalizeOutputPayload struct {
|
|
LaneID string `json:"lane_id"`
|
|
NormalizerKey string `json:"normalizer_key"`
|
|
SourceID string `json:"source_id,omitempty"`
|
|
Schema contracts.ResponseSchema `json:"schema,omitempty"`
|
|
Payload binaryEnvelope `json:"payload"`
|
|
}
|
|
|
|
type binaryEnvelope struct {
|
|
ContentBase64 string `json:"content_base64,omitempty"`
|
|
ContentDigest string `json:"content_digest,omitempty"`
|
|
MediaType string `json:"media_type,omitempty"`
|
|
Metadata map[string]any `json:"metadata,omitempty"`
|
|
Warnings []contracts.Warning `json:"warnings,omitempty"`
|
|
}
|
|
|
|
func chunkEnvelopes(chunks []contracts.SourceChunk) []chunkEnvelope {
|
|
if len(chunks) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]chunkEnvelope, 0, len(chunks))
|
|
for _, chunk := range chunks {
|
|
out = append(out, chunkEnvelope{
|
|
ID: chunk.ID,
|
|
SourceID: chunk.SourceID,
|
|
Index: chunk.Index,
|
|
StartUnitID: chunk.StartUnitID,
|
|
EndUnitID: chunk.EndUnitID,
|
|
Content: binaryEnvelopeFromContent(chunk.Content, chunk.MediaType, chunk.Metadata, nil),
|
|
Units: cloneSourceUnits(chunk.Units),
|
|
Metadata: cloneMetadata(chunk.Metadata),
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func extractOutputEnvelopes(outputs []contracts.ExtractOutput) []extractOutputEnvelope {
|
|
if len(outputs) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]extractOutputEnvelope, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
out = append(out, extractOutputEnvelope{
|
|
LaneID: output.LaneID,
|
|
ExtractorKey: output.ExtractorKey,
|
|
SourceID: output.SourceID,
|
|
ChunkID: output.ChunkID,
|
|
ChunkIndex: output.ChunkIndex,
|
|
Schema: schemaEnvelope(output.Schema),
|
|
Payload: binaryEnvelopeFromPayload(output.Payload),
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func mergeOutputEnvelopeFromOutput(output contracts.MergeOutput) mergeOutputPayload {
|
|
return mergeOutputPayload{
|
|
LaneID: output.LaneID,
|
|
MergerKey: output.MergerKey,
|
|
SourceID: output.SourceID,
|
|
Schema: schemaEnvelope(output.Schema),
|
|
Payload: binaryEnvelopeFromPayload(output.Payload),
|
|
}
|
|
}
|
|
|
|
func normalizeOutputEnvelopeFromOutput(output contracts.NormalizeOutput) normalizeOutputPayload {
|
|
return normalizeOutputPayload{
|
|
LaneID: output.LaneID,
|
|
NormalizerKey: output.NormalizerKey,
|
|
SourceID: output.SourceID,
|
|
Schema: schemaEnvelope(output.Schema),
|
|
Payload: binaryEnvelopeFromPayload(output.Payload),
|
|
}
|
|
}
|
|
|
|
func schemaEnvelope(schema contracts.ResponseSchema) contracts.ResponseSchema {
|
|
schema.JSONSchema = nil
|
|
return schema
|
|
}
|
|
|
|
func binaryEnvelopeFromPayload(payload contracts.RawPayload) binaryEnvelope {
|
|
return binaryEnvelopeFromContent(payload.Content, payload.MediaType, payload.Metadata, payload.Warnings)
|
|
}
|
|
|
|
func binaryEnvelopeFromContent(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) binaryEnvelope {
|
|
return binaryEnvelope{
|
|
ContentBase64: base64.StdEncoding.EncodeToString(content),
|
|
ContentDigest: contentDigest(content),
|
|
MediaType: mediaType,
|
|
Metadata: cloneMetadata(metadata),
|
|
Warnings: cloneWarnings(warnings),
|
|
}
|
|
}
|
|
|
|
func cloneSourceDocument(doc source.SourceDocument) source.SourceDocument {
|
|
doc.Units = cloneSourceUnits(doc.Units)
|
|
doc.Metadata = cloneMetadata(doc.Metadata)
|
|
return doc
|
|
}
|
|
|
|
func cloneSourceUnits(units []source.SourceUnit) []source.SourceUnit {
|
|
if len(units) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]source.SourceUnit, 0, len(units))
|
|
for _, unit := range units {
|
|
out = append(out, source.SourceUnit{
|
|
ID: unit.ID,
|
|
Kind: unit.Kind,
|
|
Text: unit.Text,
|
|
Metadata: cloneMetadata(unit.Metadata),
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
|
|
if len(warnings) == 0 {
|
|
return nil
|
|
}
|
|
return append([]contracts.Warning(nil), warnings...)
|
|
}
|
|
|
|
func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput {
|
|
if len(rejected) == 0 {
|
|
return nil
|
|
}
|
|
return append([]contracts.RejectedOutput(nil), rejected...)
|
|
}
|
|
|
|
func cloneMetadata(metadata map[string]any) map[string]any {
|
|
if len(metadata) == 0 {
|
|
return nil
|
|
}
|
|
out := make(map[string]any, len(metadata))
|
|
for key, value := range metadata {
|
|
out[key] = value
|
|
}
|
|
return out
|
|
}
|
|
|
|
func rawOutputDigests(payloads []contracts.RawPayload) []pipeline.CheckpointFingerprint {
|
|
values := make([]pipeline.CheckpointFingerprint, 0, len(payloads))
|
|
for i, payload := range payloads {
|
|
values = append(values, pipeline.CheckpointFingerprint{
|
|
Name: fmt.Sprintf("payload[%d]", i),
|
|
Value: contentDigest(payload.Content),
|
|
})
|
|
}
|
|
return normalizeFingerprints(values)
|
|
}
|
|
|
|
func extractPayloads(outputs []contracts.ExtractOutput) []contracts.RawPayload {
|
|
if len(outputs) == 0 {
|
|
return nil
|
|
}
|
|
payloads := make([]contracts.RawPayload, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
payloads = append(payloads, output.Payload)
|
|
}
|
|
return payloads
|
|
}
|
|
|
|
func chunkOutputDigests(chunks []contracts.SourceChunk) []pipeline.CheckpointFingerprint {
|
|
values := make([]pipeline.CheckpointFingerprint, 0, len(chunks))
|
|
for _, chunk := range chunks {
|
|
values = append(values, pipeline.CheckpointFingerprint{
|
|
Name: chunk.ID,
|
|
Value: contentDigest(chunk.Content),
|
|
})
|
|
}
|
|
return normalizeFingerprints(values)
|
|
}
|
|
|
|
func digestFingerprints(name string, digest string) []pipeline.CheckpointFingerprint {
|
|
digest = strings.TrimSpace(digest)
|
|
if digest == "" {
|
|
return nil
|
|
}
|
|
return []pipeline.CheckpointFingerprint{{Name: name, Value: digest}}
|
|
}
|
|
|
|
func workspaceFingerprints(values []pipeline.CheckpointFingerprint) []coreworkspace.Fingerprint {
|
|
normalized := normalizeFingerprints(values)
|
|
if len(normalized) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]coreworkspace.Fingerprint, 0, len(normalized))
|
|
for _, value := range normalized {
|
|
out = append(out, coreworkspace.Fingerprint{Name: value.Name, Value: value.Value})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func normalizeFingerprints(values []pipeline.CheckpointFingerprint) []pipeline.CheckpointFingerprint {
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
byName := make(map[string]string, len(values))
|
|
for _, value := range values {
|
|
name := strings.TrimSpace(value.Name)
|
|
fingerprint := strings.TrimSpace(value.Value)
|
|
if name == "" || fingerprint == "" {
|
|
continue
|
|
}
|
|
byName[name] = fingerprint
|
|
}
|
|
if len(byName) == 0 {
|
|
return nil
|
|
}
|
|
names := make([]string, 0, len(byName))
|
|
for name := range byName {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
out := make([]pipeline.CheckpointFingerprint, 0, len(names))
|
|
for _, name := range names {
|
|
out = append(out, pipeline.CheckpointFingerprint{Name: name, Value: byName[name]})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func rejectionSummaries(rejected []contracts.RejectedOutput) []coreworkspace.RejectionSummary {
|
|
if len(rejected) == 0 {
|
|
return nil
|
|
}
|
|
type key struct {
|
|
validatorName string
|
|
reasonCode string
|
|
message string
|
|
}
|
|
counts := make(map[key]int, len(rejected))
|
|
for _, item := range rejected {
|
|
k := key{validatorName: item.ValidatorName, reasonCode: item.ReasonCode, message: item.Message}
|
|
counts[k]++
|
|
}
|
|
keys := make([]key, 0, len(counts))
|
|
for k := range counts {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Slice(keys, func(i, j int) bool {
|
|
if keys[i].validatorName != keys[j].validatorName {
|
|
return keys[i].validatorName < keys[j].validatorName
|
|
}
|
|
if keys[i].reasonCode != keys[j].reasonCode {
|
|
return keys[i].reasonCode < keys[j].reasonCode
|
|
}
|
|
return keys[i].message < keys[j].message
|
|
})
|
|
out := make([]coreworkspace.RejectionSummary, 0, len(keys))
|
|
for _, k := range keys {
|
|
out = append(out, coreworkspace.RejectionSummary{
|
|
ValidatorName: k.validatorName,
|
|
ReasonCode: k.reasonCode,
|
|
Message: k.message,
|
|
Count: counts[k],
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func statusForRejected(rejected []contracts.RejectedOutput) coreworkspace.StageStatus {
|
|
if len(rejected) > 0 {
|
|
return coreworkspace.StatusSucceededWithRejections
|
|
}
|
|
return coreworkspace.StatusSucceeded
|
|
}
|
|
|
|
func validationStatusString(warnings []contracts.Warning, rejected []contracts.RejectedOutput) string {
|
|
if len(rejected) > 0 {
|
|
return "rejected"
|
|
}
|
|
if len(warnings) > 0 {
|
|
return "approved_with_warnings"
|
|
}
|
|
return "approved"
|
|
}
|
|
|
|
func errorMetadata(err error) map[string]string {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return map[string]string{"error": err.Error()}
|
|
}
|
|
|
|
func laneManifestPath(stage string, laneID string) string {
|
|
return lanePayloadPath(stage, laneID, "manifest.json")
|
|
}
|
|
|
|
func lanePayloadPath(stage string, laneID string, file string) string {
|
|
return path.Join(stage, checkpointPathComponent(laneID), file)
|
|
}
|
|
|
|
func checkpointPathComponent(value string) string {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
return "_"
|
|
}
|
|
var b strings.Builder
|
|
for _, r := range value {
|
|
switch {
|
|
case r >= 'a' && r <= 'z':
|
|
b.WriteRune(r)
|
|
case r >= 'A' && r <= 'Z':
|
|
b.WriteRune(r)
|
|
case r >= '0' && r <= '9':
|
|
b.WriteRune(r)
|
|
case r == '-' || r == '_' || r == '.':
|
|
b.WriteRune(r)
|
|
default:
|
|
b.WriteString(fmt.Sprintf("~%x", r))
|
|
}
|
|
}
|
|
out := b.String()
|
|
if out == "." || out == ".." || strings.Contains(out, "..") {
|
|
return "_"
|
|
}
|
|
return out
|
|
}
|
|
|
|
func contentDigest(content []byte) string {
|
|
sum := sha256.Sum256(content)
|
|
return "sha256:" + hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
func timePtr(t time.Time) *time.Time {
|
|
return &t
|
|
}
|