krply/internal/materialize/diff.go

package materialize

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"time"
)

// DiffResult is the semantic diff between two points in time.
type DiffResult struct {
	Before  time.Time
	After   time.Time
	Changes []ObjectDiff
	HasGaps bool
	Warning string
}

// ObjectDiff is the set of field changes for one object.
type ObjectDiff struct {
	Namespace string
	Name      string
	Kind      string
	Changes   []FieldChange
}

// FieldChange is one changed field, addressed by a dotted JSON path.
type FieldChange struct {
	Path    string
	Before  any
	After   any
	Added   bool
	Removed bool
}

// metadataIgnoreKeys are server-owned metadata fields, ignored only when they
// appear directly under metadata.
var metadataIgnoreKeys = map[string]bool{
	"uid":               true,
	"resourceVersion":   true,
	"creationTimestamp": true,
	"generation":        true,
	"managedFields":     true,
	"deletionTimestamp": true,
	"ownerReferences":   true,
}

// ignoreKey reports whether a map key at parentPath is server-owned noise.
// Ignoring is path-scoped so a user field named "status", "uid" or
// "resourceVersion" inside data or spec is never silently dropped.
func ignoreKey(parentPath, key string) bool {
	if key == "status" && parentPath == "" {
		return true
	}
	if parentPath == "metadata" && metadataIgnoreKeys[key] {
		return true
	}
	if key == "kubectl.kubernetes.io/last-applied-configuration" && parentPath == "metadata.annotations" {
		return true
	}
	return false
}

// Diff reconstructs cluster state before and after, intersects object keys,
// and reports semantic field changes, ignoring server-owned metadata.
func (m *Materializer) Diff(ctx context.Context, clusterID, namespace string, before, after time.Time) (*DiffResult, error) {
	if !before.IsZero() && !after.IsZero() && before.After(after) {
		return nil, fmt.Errorf("materialize: before (%s) must not be after after (%s)", before.Format(time.RFC3339), after.Format(time.RFC3339))
	}
	streams, err := m.store.Streams(ctx)
	if err != nil {
		return nil, err
	}

	res := &DiffResult{Before: before, After: after}
	beforeStates := map[string]ObjectState{}
	afterStates := map[string]ObjectState{}
	var gapped []string

	for _, s := range streams {
		if s.ClusterID != clusterID {
			continue
		}
		if !before.IsZero() {
			b, err := m.StreamState(ctx, s.StreamID, before)
			if err != nil {
				return nil, err
			}
			if b.HasGaps {
				gapped = append(gapped, s.StreamID)
			}
			for _, st := range b.Objects {
				if namespace != "" && st.Namespace != namespace {
					continue
				}
				beforeStates[st.StreamID+"|"+st.Namespace+"/"+st.Name] = st
			}
		}
		a, err := m.StreamState(ctx, s.StreamID, after)
		if err != nil {
			return nil, err
		}
		if a.HasGaps {
			gapped = append(gapped, s.StreamID)
		}
		for _, st := range a.Objects {
			if namespace != "" && st.Namespace != namespace {
				continue
			}
			afterStates[st.StreamID+"|"+st.Namespace+"/"+st.Name] = st
		}
	}

	keys := make([]string, 0, len(beforeStates)+len(afterStates))
	seen := map[string]bool{}
	for k := range beforeStates {
		if !seen[k] {
			seen[k] = true
			keys = append(keys, k)
		}
	}
	for k := range afterStates {
		if !seen[k] {
			seen[k] = true
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)

	for _, k := range keys {
		b, okB := beforeStates[k]
		a, okA := afterStates[k]
		od := ObjectDiff{}
		var beforeAny, afterAny any
		if okB {
			od.Namespace, od.Name, od.Kind = b.Namespace, b.Name, b.Kind
			beforeAny = parseObject(b.Object)
		}
		if okA {
			od.Namespace, od.Name, od.Kind = a.Namespace, a.Name, a.Kind
			afterAny = parseObject(a.Object)
		}
		switch {
		case okB && !okA:
			od.Changes = []FieldChange{{Path: "", Removed: true, Before: beforeAny}}
		case !okB && okA:
			od.Changes = []FieldChange{{Path: "", Added: true, After: afterAny}}
		default:
			diffValue("", beforeAny, afterAny, &od.Changes)
			sort.Slice(od.Changes, func(i, j int) bool { return od.Changes[i].Path < od.Changes[j].Path })
			if len(od.Changes) == 0 {
				continue
			}
		}
		res.Changes = append(res.Changes, od)
	}

	sort.Slice(res.Changes, func(i, j int) bool {
		if res.Changes[i].Namespace != res.Changes[j].Namespace {
			return res.Changes[i].Namespace < res.Changes[j].Namespace
		}
		if res.Changes[i].Name != res.Changes[j].Name {
			return res.Changes[i].Name < res.Changes[j].Name
		}
		return res.Changes[i].Kind < res.Changes[j].Kind
	})

	if len(gapped) > 0 {
		sort.Strings(gapped)
		res.HasGaps = true
		res.Warning = "coverage incomplete for stream(s): " + strings.Join(gapped, ", ")
	}

	return res, nil
}

// DiffObjects computes the field changes between two raw object payloads,
// ignoring server-owned metadata. It is used for timeline summaries.
func DiffObjects(before, after json.RawMessage) []FieldChange {
	var out []FieldChange
	if len(before) == 0 && len(after) == 0 {
		return out
	}
	if len(before) == 0 {
		return []FieldChange{{Path: "", Added: true, After: parseObject(after)}}
	}
	if len(after) == 0 {
		return []FieldChange{{Path: "", Removed: true, Before: parseObject(before)}}
	}
	diffValue("", parseObject(before), parseObject(after), &out)
	sort.Slice(out, func(i, j int) bool { return out[i].Path < out[j].Path })
	return out
}

// parseObject decodes a raw object into an arbitrary JSON value.
func parseObject(raw json.RawMessage) any {
	var v any
	if err := json.Unmarshal(raw, &v); err != nil {
		return string(raw)
	}
	return v
}

// diffValue recurses into before/after and appends field changes at out.
// Arrays are compared by index; a length change is reported as one change at
// the array's path. Map objects recurse by key; keys present on only one side
// are reported as a single Added/Removed change.
func diffValue(path string, before, after any, out *[]FieldChange) {
	if reflect.DeepEqual(before, after) {
		return
	}
	bm, bOK := before.(map[string]any)
	am, aOK := after.(map[string]any)
	if bOK && aOK {
		keys := map[string]bool{}
		for k := range bm {
			keys[k] = true
		}
		for k := range am {
			keys[k] = true
		}
		sorted := make([]string, 0, len(keys))
		for k := range keys {
			sorted = append(sorted, k)
		}
		sort.Strings(sorted)
		for _, k := range sorted {
			if ignoreKey(path, k) {
				continue
			}
			childPath := joinPath(path, k)
			_, beforeHas := bm[k]
			_, afterHas := am[k]
			switch {
			case beforeHas && !afterHas:
				*out = append(*out, FieldChange{Path: childPath, Before: bm[k], Removed: true})
			case !beforeHas && afterHas:
				*out = append(*out, FieldChange{Path: childPath, After: am[k], Added: true})
			default:
				diffValue(childPath, bm[k], am[k], out)
			}
		}
		return
	}
	bl, bArr := before.([]any)
	al, aArr := after.([]any)
	if bArr && aArr {
		if len(bl) != len(al) {
			*out = append(*out, FieldChange{Path: path, Before: before, After: after})
			return
		}
		for i := range bl {
			diffValue(fmt.Sprintf("%s[%d]", path, i), bl[i], al[i], out)
		}
		return
	}
	*out = append(*out, FieldChange{Path: path, Before: before, After: after})
}

// joinPath builds a dotted path. Segments that could collide with the path
// syntax ('.', '[', ']', '\') are backslash-escaped.
func joinPath(parent, key string) string {
	key = escapePathSegment(key)
	if parent == "" {
		return key
	}
	return parent + "." + key
}

func escapePathSegment(k string) string {
	k = strings.ReplaceAll(k, `\`, `\\`)
	k = strings.ReplaceAll(k, `.`, `\.`)
	k = strings.ReplaceAll(k, `[`, `\[`)
	k = strings.ReplaceAll(k, `]`, `\]`)
	return k
}