372 lines
11 KiB
Go
372 lines
11 KiB
Go
package state
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
)
|
|
|
|
const (
|
|
SchemaVersion = 2
|
|
SharedRootSchemaVersion = 3
|
|
legacySchemaVersion = 1
|
|
StateModeSingleOwner = config.StateModeSingleOwner
|
|
StateModeSharedRoot = config.StateModeSharedRoot
|
|
)
|
|
|
|
type DistributorState struct {
|
|
SchemaVersion int
|
|
DistributorVersion string
|
|
PipelineID string
|
|
DestinationID string
|
|
PublishedAt time.Time
|
|
CreatedAt time.Time
|
|
UpdatedAt time.Time
|
|
State StatePolicy
|
|
Reconciliation ReconciliationPolicy
|
|
Source SourceState
|
|
Links *LinkState
|
|
Outputs []OutputFile
|
|
}
|
|
|
|
type StatePolicy struct {
|
|
Mode string
|
|
}
|
|
|
|
type ReconciliationPolicy struct {
|
|
Mode string
|
|
}
|
|
|
|
type SourceState struct {
|
|
Manifest bundle.Manifest
|
|
}
|
|
|
|
type LinkState struct {
|
|
PrimaryURL string
|
|
}
|
|
|
|
type OutputFile struct {
|
|
Path string
|
|
Kind string
|
|
SourcePath string
|
|
Transform string
|
|
URL string
|
|
SHA256 string
|
|
Size int64
|
|
CreatedAt time.Time
|
|
UpdatedAt time.Time
|
|
}
|
|
|
|
type rawDistributorState struct {
|
|
SchemaVersion *int `json:"schema_version"`
|
|
DistributorVersion string `json:"distributor_version"`
|
|
PipelineID *string `json:"pipeline_id"`
|
|
DestinationID *string `json:"destination_id"`
|
|
PublishedAt *string `json:"published_at"`
|
|
CreatedAt *string `json:"created_at"`
|
|
UpdatedAt *string `json:"updated_at"`
|
|
State *rawStatePolicy `json:"state"`
|
|
Reconciliation *rawReconciliationPolicy `json:"reconciliation"`
|
|
Source *rawSourceState `json:"source"`
|
|
Links *rawLinkState `json:"links"`
|
|
Outputs []rawOutputFile `json:"outputs"`
|
|
}
|
|
|
|
type rawStatePolicy struct {
|
|
Mode string `json:"mode"`
|
|
}
|
|
|
|
type rawReconciliationPolicy struct {
|
|
Mode string `json:"mode"`
|
|
}
|
|
|
|
type rawSourceState struct {
|
|
Manifest json.RawMessage `json:"manifest"`
|
|
}
|
|
|
|
type rawLinkState struct {
|
|
PrimaryURL string `json:"primary_url"`
|
|
}
|
|
|
|
type rawOutputFile struct {
|
|
Path *string `json:"path"`
|
|
Kind *string `json:"kind"`
|
|
SourcePath *string `json:"source_path"`
|
|
Transform string `json:"transform"`
|
|
URL string `json:"url"`
|
|
SHA256 *string `json:"sha256"`
|
|
Size *int64 `json:"size"`
|
|
CreatedAt *string `json:"created_at"`
|
|
UpdatedAt *string `json:"updated_at"`
|
|
}
|
|
|
|
func Parse(data []byte) (DistributorState, error) {
|
|
decoder := json.NewDecoder(bytes.NewReader(data))
|
|
var raw rawDistributorState
|
|
if err := decoder.Decode(&raw); err != nil {
|
|
return DistributorState{}, fmt.Errorf("parse distributor state: %w", err)
|
|
}
|
|
var extra any
|
|
if err := decoder.Decode(&extra); err != io.EOF {
|
|
return DistributorState{}, fmt.Errorf("parse distributor state: trailing data")
|
|
}
|
|
state, err := parseRaw(raw)
|
|
if err != nil {
|
|
return DistributorState{}, err
|
|
}
|
|
if err := Validate(state); err != nil {
|
|
return DistributorState{}, err
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
func parseRaw(raw rawDistributorState) (DistributorState, error) {
|
|
var state DistributorState
|
|
if raw.SchemaVersion == nil {
|
|
return DistributorState{}, fmt.Errorf("state schema_version is required")
|
|
}
|
|
state.SchemaVersion = *raw.SchemaVersion
|
|
if state.SchemaVersion != SchemaVersion && state.SchemaVersion != legacySchemaVersion {
|
|
return DistributorState{}, fmt.Errorf("state schema_version must be %d or %d", legacySchemaVersion, SchemaVersion)
|
|
}
|
|
legacy := state.SchemaVersion == legacySchemaVersion
|
|
state.DistributorVersion = raw.DistributorVersion
|
|
if raw.PipelineID == nil || *raw.PipelineID == "" {
|
|
return DistributorState{}, fmt.Errorf("state pipeline_id is required")
|
|
}
|
|
state.PipelineID = *raw.PipelineID
|
|
if raw.DestinationID == nil || *raw.DestinationID == "" {
|
|
return DistributorState{}, fmt.Errorf("state destination_id is required")
|
|
}
|
|
state.DestinationID = *raw.DestinationID
|
|
if raw.PublishedAt == nil || *raw.PublishedAt == "" {
|
|
return DistributorState{}, fmt.Errorf("state published_at is required")
|
|
}
|
|
publishedAt, err := time.Parse(time.RFC3339, *raw.PublishedAt)
|
|
if err != nil {
|
|
return DistributorState{}, fmt.Errorf("state published_at must be RFC3339: %w", err)
|
|
}
|
|
state.PublishedAt = publishedAt.UTC()
|
|
if legacy {
|
|
state.SchemaVersion = SchemaVersion
|
|
state.CreatedAt = state.PublishedAt
|
|
state.UpdatedAt = state.PublishedAt
|
|
state.State.Mode = StateModeSingleOwner
|
|
state.Reconciliation.Mode = config.ReconciliationModeReplace
|
|
} else {
|
|
createdAt, err := parseRequiredTime("state created_at", raw.CreatedAt)
|
|
if err != nil {
|
|
return DistributorState{}, err
|
|
}
|
|
updatedAt, err := parseRequiredTime("state updated_at", raw.UpdatedAt)
|
|
if err != nil {
|
|
return DistributorState{}, err
|
|
}
|
|
state.CreatedAt = createdAt
|
|
state.UpdatedAt = updatedAt
|
|
if raw.State == nil || raw.State.Mode == "" {
|
|
return DistributorState{}, fmt.Errorf("state state.mode is required")
|
|
}
|
|
state.State.Mode = raw.State.Mode
|
|
if raw.Reconciliation == nil || raw.Reconciliation.Mode == "" {
|
|
return DistributorState{}, fmt.Errorf("state reconciliation.mode is required")
|
|
}
|
|
state.Reconciliation.Mode = raw.Reconciliation.Mode
|
|
}
|
|
if raw.Source == nil || len(raw.Source.Manifest) == 0 {
|
|
return DistributorState{}, fmt.Errorf("state source.manifest is required")
|
|
}
|
|
manifest, err := bundle.ParseManifest(raw.Source.Manifest)
|
|
if err != nil {
|
|
return DistributorState{}, fmt.Errorf("state source.manifest: %w", err)
|
|
}
|
|
state.Source.Manifest = manifest
|
|
if raw.Links != nil {
|
|
state.Links = &LinkState{PrimaryURL: raw.Links.PrimaryURL}
|
|
}
|
|
if raw.Outputs == nil {
|
|
return DistributorState{}, fmt.Errorf("state outputs is required")
|
|
}
|
|
outputs, err := parseOutputs(raw.Outputs, legacy, state.PublishedAt)
|
|
if err != nil {
|
|
return DistributorState{}, err
|
|
}
|
|
state.Outputs = outputs
|
|
return state, nil
|
|
}
|
|
|
|
func parseRequiredTime(context string, raw *string) (time.Time, error) {
|
|
if raw == nil || *raw == "" {
|
|
return time.Time{}, fmt.Errorf("%s is required", context)
|
|
}
|
|
parsed, err := time.Parse(time.RFC3339, *raw)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("%s must be RFC3339: %w", context, err)
|
|
}
|
|
return parsed.UTC(), nil
|
|
}
|
|
|
|
func parseOutputs(rawOutputs []rawOutputFile, legacy bool, publishedAt time.Time) ([]OutputFile, error) {
|
|
outputs := make([]OutputFile, 0, len(rawOutputs))
|
|
seen := make(map[string]struct{}, len(rawOutputs))
|
|
for index, raw := range rawOutputs {
|
|
output, err := parseOutput(index, raw, legacy, publishedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, exists := seen[output.Path]; exists {
|
|
return nil, fmt.Errorf("state outputs[%d].path duplicates %q", index, output.Path)
|
|
}
|
|
seen[output.Path] = struct{}{}
|
|
outputs = append(outputs, output)
|
|
}
|
|
return outputs, nil
|
|
}
|
|
|
|
func parseOutput(index int, raw rawOutputFile, legacy bool, publishedAt time.Time) (OutputFile, error) {
|
|
if raw.Path == nil || *raw.Path == "" {
|
|
return OutputFile{}, fmt.Errorf("state outputs[%d].path is required", index)
|
|
}
|
|
if raw.Kind == nil || *raw.Kind == "" {
|
|
return OutputFile{}, fmt.Errorf("state outputs[%d].kind is required", index)
|
|
}
|
|
if raw.SourcePath == nil || *raw.SourcePath == "" {
|
|
return OutputFile{}, fmt.Errorf("state outputs[%d].source_path is required", index)
|
|
}
|
|
if raw.SHA256 == nil || *raw.SHA256 == "" {
|
|
return OutputFile{}, fmt.Errorf("state outputs[%d].sha256 is required", index)
|
|
}
|
|
if raw.Size == nil {
|
|
return OutputFile{}, fmt.Errorf("state outputs[%d].size is required", index)
|
|
}
|
|
createdAt := publishedAt
|
|
updatedAt := publishedAt
|
|
if !legacy {
|
|
var err error
|
|
createdAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].created_at", index), raw.CreatedAt)
|
|
if err != nil {
|
|
return OutputFile{}, err
|
|
}
|
|
updatedAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].updated_at", index), raw.UpdatedAt)
|
|
if err != nil {
|
|
return OutputFile{}, err
|
|
}
|
|
}
|
|
return OutputFile{
|
|
Path: *raw.Path,
|
|
Kind: *raw.Kind,
|
|
SourcePath: *raw.SourcePath,
|
|
Transform: raw.Transform,
|
|
URL: raw.URL,
|
|
SHA256: *raw.SHA256,
|
|
Size: *raw.Size,
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: updatedAt,
|
|
}, nil
|
|
}
|
|
|
|
func (s DistributorState) PublishedAtString() string {
|
|
return s.PublishedAt.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func (s DistributorState) CreatedAtString() string {
|
|
return s.CreatedAt.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func (s DistributorState) UpdatedAtString() string {
|
|
return s.UpdatedAt.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func (o OutputFile) CreatedAtString() string {
|
|
return o.CreatedAt.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func (o OutputFile) UpdatedAtString() string {
|
|
return o.UpdatedAt.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func (s DistributorState) MarshalJSON() ([]byte, error) {
|
|
type sourceJSON struct {
|
|
Manifest bundle.Manifest `json:"manifest"`
|
|
}
|
|
type stateJSON struct {
|
|
SchemaVersion int `json:"schema_version"`
|
|
DistributorVersion string `json:"distributor_version,omitempty"`
|
|
PipelineID string `json:"pipeline_id"`
|
|
DestinationID string `json:"destination_id"`
|
|
PublishedAt string `json:"published_at"`
|
|
CreatedAt string `json:"created_at"`
|
|
UpdatedAt string `json:"updated_at"`
|
|
State StatePolicy `json:"state"`
|
|
Reconciliation ReconciliationPolicy `json:"reconciliation"`
|
|
Source sourceJSON `json:"source"`
|
|
Links *LinkState `json:"links,omitempty"`
|
|
Outputs []OutputFile `json:"outputs"`
|
|
}
|
|
return json.Marshal(stateJSON{
|
|
SchemaVersion: s.SchemaVersion,
|
|
DistributorVersion: s.DistributorVersion,
|
|
PipelineID: s.PipelineID,
|
|
DestinationID: s.DestinationID,
|
|
PublishedAt: s.PublishedAtString(),
|
|
CreatedAt: s.CreatedAtString(),
|
|
UpdatedAt: s.UpdatedAtString(),
|
|
State: s.State,
|
|
Reconciliation: s.Reconciliation,
|
|
Source: sourceJSON{Manifest: s.Source.Manifest},
|
|
Links: s.Links,
|
|
Outputs: s.Outputs,
|
|
})
|
|
}
|
|
|
|
func (p StatePolicy) MarshalJSON() ([]byte, error) {
|
|
type policyJSON struct {
|
|
Mode string `json:"mode"`
|
|
}
|
|
return json.Marshal(policyJSON{Mode: p.Mode})
|
|
}
|
|
|
|
func (p ReconciliationPolicy) MarshalJSON() ([]byte, error) {
|
|
type policyJSON struct {
|
|
Mode string `json:"mode"`
|
|
}
|
|
return json.Marshal(policyJSON{Mode: p.Mode})
|
|
}
|
|
|
|
func (l LinkState) MarshalJSON() ([]byte, error) {
|
|
type linkJSON struct {
|
|
PrimaryURL string `json:"primary_url,omitempty"`
|
|
}
|
|
return json.Marshal(linkJSON{PrimaryURL: l.PrimaryURL})
|
|
}
|
|
|
|
func (o OutputFile) MarshalJSON() ([]byte, error) {
|
|
type outputJSON struct {
|
|
Path string `json:"path"`
|
|
Kind string `json:"kind"`
|
|
SourcePath string `json:"source_path"`
|
|
Transform string `json:"transform,omitempty"`
|
|
URL string `json:"url,omitempty"`
|
|
SHA256 string `json:"sha256"`
|
|
Size int64 `json:"size"`
|
|
CreatedAt string `json:"created_at"`
|
|
UpdatedAt string `json:"updated_at"`
|
|
}
|
|
return json.Marshal(outputJSON{
|
|
Path: o.Path,
|
|
Kind: o.Kind,
|
|
SourcePath: o.SourcePath,
|
|
Transform: o.Transform,
|
|
URL: o.URL,
|
|
SHA256: o.SHA256,
|
|
Size: o.Size,
|
|
CreatedAt: o.CreatedAt.UTC().Format(time.RFC3339),
|
|
UpdatedAt: o.UpdatedAt.UTC().Format(time.RFC3339),
|
|
})
|
|
}
|