Remove legacy publish and state paths

internal/state/document.go

@@ -0,0 +1,107 @@
+package state
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"time"
+)
+
+const (
+	legacyOwnerSchema      = 2
+	legacyMultiOwnerSchema = 3
+	CatalogSchemaVersion   = 4
+	legacySchemaVersion    = 1
+	StateModeCatalog       = "catalog"
+	OutputKindSource       = "source"
+	OutputKindGenerated    = "generated"
+)
+
+type StateDocument struct {
+	Catalog          *CatalogState
+	SupersededLegacy *SupersededLegacyState
+}
+
+type DestinationStatus struct {
+	Catalog          *CatalogState
+	SupersededLegacy *SupersededLegacyState
+	StateErr         error
+	HasContents      bool
+}
+
+type SupersededLegacyState struct {
+	SchemaVersion int
+}
+
+type StatePolicy struct {
+	Mode string
+}
+
+type OwnerScope struct {
+	PipelineID    string
+	DestinationID string
+}
+
+type rawStatePolicy struct {
+	Mode string `json:"mode"`
+}
+
+func ParseDocument(data []byte) (StateDocument, error) {
+	schemaVersion, err := parseSchemaVersion(data)
+	if err != nil {
+		return StateDocument{}, err
+	}
+	switch schemaVersion {
+	case legacySchemaVersion, legacyOwnerSchema, legacyMultiOwnerSchema:
+		return StateDocument{SupersededLegacy: &SupersededLegacyState{SchemaVersion: schemaVersion}}, nil
+	case CatalogSchemaVersion:
+		catalog, err := ParseCatalog(data)
+		if err != nil {
+			return StateDocument{}, err
+		}
+		return StateDocument{Catalog: &catalog}, nil
+	default:
+		return StateDocument{}, fmt.Errorf("state schema_version %d is unsupported", schemaVersion)
+	}
+}
+
+func parseSchemaVersion(data []byte) (int, error) {
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	var raw struct {
+		SchemaVersion *int `json:"schema_version"`
+	}
+	if err := decoder.Decode(&raw); err != nil {
+		return 0, fmt.Errorf("parse distributor state: %w", err)
+	}
+	var extra any
+	if err := decoder.Decode(&extra); err != io.EOF {
+		return 0, fmt.Errorf("parse distributor state: trailing data")
+	}
+	if raw.SchemaVersion == nil {
+		return 0, fmt.Errorf("state schema_version is required")
+	}
+	return *raw.SchemaVersion, nil
+}
+
+func (p StatePolicy) MarshalJSON() ([]byte, error) {
+	type policyJSON struct {
+		Mode string `json:"mode"`
+	}
+	return json.Marshal(policyJSON{Mode: p.Mode})
+}
+
+func CurrentOwnerScope(pipelineID, destinationID string) OwnerScope {
+	return OwnerScope{PipelineID: pipelineID, DestinationID: destinationID}
+}
+
+func parseRequiredTime(field string, raw *string) (time.Time, error) {
+	if raw == nil || *raw == "" {
+		return time.Time{}, fmt.Errorf("%s is required", field)
+	}
+	parsed, err := time.Parse(time.RFC3339, *raw)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("%s must be RFC3339: %w", field, err)
+	}
+	return parsed.UTC(), nil
+}