notarius/internal/framework/checkpoint/loader.go

package checkpoint

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"os"
	"strings"

	"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)

type FilesystemLoader struct {
	root           string
	identityDigest string
}

func NewFilesystemLoader(root string, identity Identity) (pipeline.CheckpointLoader, error) {
	root = strings.TrimSpace(root)
	if root == "" {
		return pipeline.NoopCheckpointLoader(), nil
	}
	relative, err := identity.RelativePath()
	if err != nil {
		return nil, err
	}
	target, err := fileio.SafePath(root, relative)
	if err != nil {
		return nil, err
	}
	return &FilesystemLoader{root: target, identityDigest: identity.Digest}, nil
}

func (l *FilesystemLoader) Enabled() bool {
	return l != nil && strings.TrimSpace(l.root) != ""
}

func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) {
	var manifest SourceManifest
	if decision := l.readJSON("source/manifest.json", &manifest); !decision.Reused {
		return pipeline.SourceCheckpoint{}, decision
	}
	if decision := l.validateManifest(manifest.StageManifest, StageSource, "", moduleKey, StatusSucceeded, nil); !decision.Reused {
		return pipeline.SourceCheckpoint{}, decision
	}
	var payload sourceDocumentEnvelope
	if decision := l.readJSON("source/source-document.json", &payload); !decision.Reused {
		return pipeline.SourceCheckpoint{}, decision
	}
	doc := cloneSourceDocument(payload.Document)
	if err := source.ValidateDocument(&doc); err != nil {
		return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint document is invalid")
	}
	if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID {
		return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint identity does not match its payload")
	}
	if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) {
		return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "source checkpoint output digest does not match its payload")
	}
	return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision()
}

func (l *FilesystemLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) {
	return l.ExtractForStep("", laneID, moduleKey, dependencies)
}

func (l *FilesystemLoader) ExtractForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) {
	var manifest ExtractLaneManifest
	if d := l.readJSON(laneManifestPath("extract", stepID, laneID), &manifest); !d.Reused {
		return pipeline.ExtractCheckpoint{}, d
	}
	if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, stepID, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused {
		return pipeline.ExtractCheckpoint{}, d
	}
	var payload artifactExtractEnvelope
	if d := l.readJSON(lanePayloadPath("extract", stepID, laneID, "outputs.json"), &payload); !d.Reused {
		return pipeline.ExtractCheckpoint{}, d
	}
	outputs, err := artifactCheckpointOutputs(payload.Outputs)
	if err != nil {
		return pipeline.ExtractCheckpoint{}, artifactDecision(err, "extract checkpoint artifact is invalid")
	}
	if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) {
		return pipeline.ExtractCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "extract checkpoint output digest does not match its payload")
	}
	return pipeline.ExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
}

func (l *FilesystemLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) {
	return l.MergeForStep("", laneID, moduleKey, dependencies)
}

func (l *FilesystemLoader) MergeForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) {
	var manifest MergeLaneManifest
	if d := l.readJSON(laneManifestPath("merge", stepID, laneID), &manifest); !d.Reused {
		return pipeline.MergeCheckpoint{}, d
	}
	if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, stepID, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
		return pipeline.MergeCheckpoint{}, d
	}
	var payload artifactSingleEnvelope
	if d := l.readJSON(lanePayloadPath("merge", stepID, laneID, "output.json"), &payload); !d.Reused {
		return pipeline.MergeCheckpoint{}, d
	}
	values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output})
	if err != nil {
		return pipeline.MergeCheckpoint{}, artifactDecision(err, "merge checkpoint artifact is invalid")
	}
	if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
		return pipeline.MergeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "merge checkpoint output digest does not match its payload")
	}
	return pipeline.MergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
}

func (l *FilesystemLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) {
	return l.NormalizeForStep("", laneID, moduleKey, dependencies)
}

func (l *FilesystemLoader) NormalizeForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) {
	var manifest NormalizeLaneManifest
	if d := l.readJSON(laneManifestPath("normalize", stepID, laneID), &manifest); !d.Reused {
		return pipeline.NormalizeCheckpoint{}, d
	}
	if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, stepID, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
		return pipeline.NormalizeCheckpoint{}, d
	}
	var payload artifactSingleEnvelope
	if d := l.readJSON(lanePayloadPath("normalize", stepID, laneID, "output.json"), &payload); !d.Reused {
		return pipeline.NormalizeCheckpoint{}, d
	}
	values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output})
	if err != nil {
		return pipeline.NormalizeCheckpoint{}, artifactDecision(err, "normalize checkpoint artifact is invalid")
	}
	if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
		return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "normalize checkpoint output digest does not match its payload")
	}
	return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
}

func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline.CheckpointArtifact, error) {
	if len(values) == 0 {
		return nil, nil
	}
	out := make([]pipeline.CheckpointArtifact, 0, len(values))
	for _, v := range values {
		content, err := contentFromEnvelope(v.Content)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(string(v.Kind)) == "" || strings.TrimSpace(v.Schema.ID) == "" || strings.TrimSpace(v.Schema.Version) == "" || strings.TrimSpace(v.SchemaDigest) == "" {
			return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactCodecIncompatible, err: fmt.Errorf("artifact codec identity is incomplete")}
		}
		out = append(out, pipeline.CheckpointArtifact{LaneID: v.LaneID, ModuleKey: v.ModuleKey, SourceID: v.SourceID, ChunkID: v.ChunkID, ChunkIndex: v.ChunkIndex, ChunkRef: v.ChunkRef, SchemaDigest: v.SchemaDigest, Artifact: contracts.SerializedArtifact{Kind: v.Kind, Schema: v.Schema, MediaType: v.Content.MediaType, Content: content, Metadata: cloneMetadata(v.Content.Metadata)}})
	}
	return out, nil
}

func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision {
	if !l.Enabled() {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled, "checkpoint loading disabled")
	}
	target, err := fileio.SafePath(l.root, name)
	if err != nil {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonPathInvalid, "checkpoint path is invalid")
	}
	data, err := os.ReadFile(target)
	if err != nil {
		if os.IsNotExist(err) {
			return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing, "checkpoint artifact is missing")
		}
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonReadFailed, "checkpoint artifact could not be read")
	}
	if err := json.Unmarshal(data, out); err != nil {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed, "checkpoint artifact could not be decoded")
	}
	return reusedDecision()
}

func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, status StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision {
	return l.validateLaneManifest(manifest, stage, "", laneID, moduleKey, dependencies, status)
}

func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, stepID string, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision {
	if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV1 {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible")
	}
	if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV2 {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible")
	}
	if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible")
	}
	if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch, "checkpoint identity does not match the current invocation")
	}
	if manifest.Stage != stage {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch, "checkpoint stage does not match the requested stage")
	}
	if strings.TrimSpace(stepID) != "" && manifest.StepID != stepID {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch, "checkpoint step does not match the requested step")
	}
	if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch, "checkpoint lane does not match the requested lane")
	}
	if strings.TrimSpace(moduleKey) != "" && manifest.ModuleKey != moduleKey {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch, "checkpoint module does not match the requested module")
	}
	statusOK := false
	for _, status := range statuses {
		if manifest.Status == status {
			statusOK = true
			break
		}
	}
	if !statusOK {
		return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable, "checkpoint status cannot be reused")
	}
	if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) {
		return decision(pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch, "checkpoint dependencies do not match")
	}
	return reusedDecision()
}

func contentFromEnvelope(value binaryEnvelope) ([]byte, error) {
	content, err := base64.StdEncoding.DecodeString(value.ContentBase64)
	if err != nil {
		return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactPayloadInvalid, err: fmt.Errorf("decode content_base64: %w", err)}
	}
	if digest := strings.TrimSpace(value.ContentDigest); digest != "" && digest != contentDigest(content) {
		return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactDigestMismatch, err: fmt.Errorf("content digest mismatch")}
	}
	return content, nil
}

func checkpointToPipelineFingerprints(values []Fingerprint) []pipeline.CheckpointFingerprint {
	if len(values) == 0 {
		return nil
	}
	out := make([]pipeline.CheckpointFingerprint, 0, len(values))
	for _, value := range values {
		out = append(out, pipeline.CheckpointFingerprint{Name: value.Name, Value: value.Value})
	}
	return normalizeFingerprints(out)
}

func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.CheckpointFingerprint) bool {
	a = normalizeFingerprints(a)
	b = normalizeFingerprints(b)
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func reusedDecision() pipeline.CheckpointDecision {
	return decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused, "checkpoint is reusable")
}

func decision(category pipeline.CheckpointDecisionCategory, code pipeline.CheckpointReasonCode, detail string) pipeline.CheckpointDecision {
	return pipeline.NewCheckpointDecision(category, code, detail)
}

type artifactPayloadError struct {
	code pipeline.CheckpointReasonCode
	err  error
}

func (e *artifactPayloadError) Error() string { return e.err.Error() }

func artifactDecision(err error, detail string) pipeline.CheckpointDecision {
	code := pipeline.CheckpointReasonArtifactPayloadInvalid
	if payloadErr, ok := err.(*artifactPayloadError); ok {
		code = payloadErr.code
	}
	return decision(pipeline.CheckpointDecisionExecuted, code, detail)
}