notarius/internal/framework/semanticreconcile/application.go

package semanticreconcile

import (
	"fmt"
	"sort"
	"strings"
)

// CloneValueFunc returns a value that shares no caller-owned mutable state with
// its input.
type CloneValueFunc[T any] func(T) T

// Record owns one typed value and its deterministic input provenance.
type Record[T any] struct {
	value                 T
	originalInputIndexes  []int
	earliestInputPosition int
	cloneValue            CloneValueFunc[T]
}

// NewRecord constructs an owned typed record. Original input indexes are
// normalized into ascending unique order.
func NewRecord[T any](value T, originalInputIndexes []int, earliestInputPosition int, cloneValue CloneValueFunc[T]) (Record[T], error) {
	if cloneValue == nil {
		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: clone value function must not be nil")
	}
	if earliestInputPosition < 0 {
		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: earliest input position must not be negative")
	}
	indexes, err := normalizeInputIndexes(originalInputIndexes)
	if err != nil {
		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: %w", err)
	}
	return Record[T]{
		value:                 cloneValue(value),
		originalInputIndexes:  indexes,
		earliestInputPosition: earliestInputPosition,
		cloneValue:            cloneValue,
	}, nil
}

// Value returns an independently owned typed value.
func (record Record[T]) Value() T {
	if record.cloneValue == nil {
		var zero T
		return zero
	}
	return record.cloneValue(record.value)
}

// OriginalInputIndexes returns an owned ascending unique index list.
func (record Record[T]) OriginalInputIndexes() []int {
	return cloneSlice(record.originalInputIndexes)
}

// EarliestInputPosition returns the earliest deterministic input position
// contributing to this record.
func (record Record[T]) EarliestInputPosition() int { return record.earliestInputPosition }

// RejectionCategory is a stable, domain-neutral reason that an otherwise safe
// semantic group was not applied.
type RejectionCategory string

// ApplicationPolicy supplies only the typed behavior needed to apply a safe
// reconciliation plan. An empty category from RejectGroup accepts the group.
type ApplicationPolicy[T any] struct {
	CloneValue       CloneValueFunc[T]
	RejectGroup      func(members []T, canonical T) RejectionCategory
	ConsolidateGroup func(members []T, canonical T) (T, error)
}

// GroupProvenance identifies the complete input contribution of one plan
// group without prescribing domain diagnostic or retry policy.
type GroupProvenance struct {
	memberPositions      []int
	canonicalPosition    int
	originalInputIndexes []int
	earliestPosition     int
}

// MemberPositions returns owned record positions in ascending order.
func (provenance GroupProvenance) MemberPositions() []int {
	return cloneSlice(provenance.memberPositions)
}

// CanonicalPosition returns the plan-selected canonical record position.
func (provenance GroupProvenance) CanonicalPosition() int {
	return provenance.canonicalPosition
}

// OriginalInputIndexes returns the sorted union contributed by all members.
func (provenance GroupProvenance) OriginalInputIndexes() []int {
	return cloneSlice(provenance.originalInputIndexes)
}

// EarliestInputPosition returns the earliest member input position.
func (provenance GroupProvenance) EarliestInputPosition() int {
	return provenance.earliestPosition
}

// AppliedGroup records the provenance of one successfully consolidated group.
type AppliedGroup struct {
	provenance GroupProvenance
}

// Provenance returns an independently owned provenance snapshot.
func (event AppliedGroup) Provenance() GroupProvenance {
	return cloneGroupProvenance(event.provenance)
}

// RejectedGroup records a typed guard decision while leaving diagnostic and retry
// construction to the consuming domain.
type RejectedGroup struct {
	category   RejectionCategory
	provenance GroupProvenance
}

// Category returns the stable neutral rejection category.
func (event RejectedGroup) Category() RejectionCategory { return event.category }

// Provenance returns an independently owned provenance snapshot.
func (event RejectedGroup) Provenance() GroupProvenance {
	return cloneGroupProvenance(event.provenance)
}

// ApplicationResult owns the ordered records and neutral events from one plan
// application.
type ApplicationResult[T any] struct {
	records        []Record[T]
	appliedGroups  []AppliedGroup
	rejectedGroups []RejectedGroup
}

// Records returns independently owned records in earliest-contribution order.
func (result ApplicationResult[T]) Records() []Record[T] {
	return cloneRecords(result.records)
}

// AppliedGroups returns independently owned applied-group events.
func (result ApplicationResult[T]) AppliedGroups() []AppliedGroup {
	events := make([]AppliedGroup, len(result.appliedGroups))
	for index, event := range result.appliedGroups {
		events[index] = AppliedGroup{provenance: cloneGroupProvenance(event.provenance)}
	}
	return preserveEmptySlice(result.appliedGroups, events)
}

// RejectedGroups returns independently owned rejected-group events.
func (result ApplicationResult[T]) RejectedGroups() []RejectedGroup {
	events := make([]RejectedGroup, len(result.rejectedGroups))
	for index, event := range result.rejectedGroups {
		events[index] = RejectedGroup{category: event.category, provenance: cloneGroupProvenance(event.provenance)}
	}
	return preserveEmptySlice(result.rejectedGroups, events)
}

type applicationEntry[T any] struct {
	record Record[T]
	order  int
}

// ApplyPlan applies safe, non-overlapping groups without mutating the plan,
// records, or values supplied to policy callbacks.
func ApplyPlan[T any](plan Plan, records []Record[T], policy ApplicationPolicy[T]) (ApplicationResult[T], error) {
	if err := validateApplicationPolicy(policy); err != nil {
		return ApplicationResult[T]{}, err
	}
	for index, record := range records {
		if err := validateRecord(record); err != nil {
			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation plan: record %d: %w", index, err)
		}
	}

	groups := plan.Groups()
	groupByFirstMember, err := validateApplicationPlan(groups, len(records))
	if err != nil {
		return ApplicationResult[T]{}, err
	}
	groupedPositions := make(map[int]struct{}, len(records))
	for _, group := range groups {
		for _, position := range group.memberPositions {
			groupedPositions[position] = struct{}{}
		}
	}

	entries := make([]applicationEntry[T], 0, len(records))
	result := ApplicationResult[T]{}
	for position, record := range records {
		group, firstMember := groupByFirstMember[position]
		if !firstMember {
			if _, grouped := groupedPositions[position]; grouped {
				continue
			}
			entries = append(entries, applicationEntry[T]{record: cloneRecordWith(record, policy.CloneValue), order: position})
			continue
		}

		provenance := groupProvenance(group, records)
		guardMembers, guardCanonical := policyInputs(group, records, policy.CloneValue)
		category := RejectionCategory("")
		if policy.RejectGroup != nil {
			category = policy.RejectGroup(guardMembers, guardCanonical)
		}
		if category != "" && strings.TrimSpace(string(category)) == "" {
			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: rejection category must not be blank", position)
		}
		if category != "" {
			for _, memberPosition := range group.memberPositions {
				entries = append(entries, applicationEntry[T]{record: cloneRecordWith(records[memberPosition], policy.CloneValue), order: memberPosition})
			}
			result.rejectedGroups = append(result.rejectedGroups, RejectedGroup{category: category, provenance: provenance})
			continue
		}

		members, canonical := policyInputs(group, records, policy.CloneValue)
		consolidated, err := policy.ConsolidateGroup(members, canonical)
		if err != nil {
			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: consolidate: %w", position, err)
		}
		owned, err := NewRecord(consolidated, provenance.originalInputIndexes, provenance.earliestPosition, policy.CloneValue)
		if err != nil {
			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: own consolidated record: %w", position, err)
		}
		entries = append(entries, applicationEntry[T]{record: owned, order: position})
		result.appliedGroups = append(result.appliedGroups, AppliedGroup{provenance: provenance})
	}

	sort.SliceStable(entries, func(left, right int) bool {
		if entries[left].record.earliestInputPosition == entries[right].record.earliestInputPosition {
			return entries[left].order < entries[right].order
		}
		return entries[left].record.earliestInputPosition < entries[right].record.earliestInputPosition
	})
	if records != nil {
		result.records = make([]Record[T], len(entries))
		for index, entry := range entries {
			result.records[index] = cloneRecord(entry.record)
		}
	}
	return result, nil
}

func validateApplicationPolicy[T any](policy ApplicationPolicy[T]) error {
	if policy.CloneValue == nil {
		return fmt.Errorf("apply semantic reconciliation plan: clone value function must not be nil")
	}
	if policy.ConsolidateGroup == nil {
		return fmt.Errorf("apply semantic reconciliation plan: consolidate group function must not be nil")
	}
	return nil
}

func validateRecord[T any](record Record[T]) error {
	if record.cloneValue == nil {
		return fmt.Errorf("invalid construction state: clone value function must not be nil")
	}
	if record.earliestInputPosition < 0 {
		return fmt.Errorf("invalid construction state: earliest input position must not be negative")
	}
	for index, value := range record.originalInputIndexes {
		if value < 0 {
			return fmt.Errorf("invalid construction state: original input index must not be negative")
		}
		if index > 0 && record.originalInputIndexes[index-1] >= value {
			return fmt.Errorf("invalid construction state: original input indexes must be ascending and unique")
		}
	}
	return nil
}

func validateApplicationPlan(groups []PlanGroup, recordCount int) (map[int]PlanGroup, error) {
	groupByFirstMember := make(map[int]PlanGroup, len(groups))
	used := make(map[int]struct{})
	for groupIndex, group := range groups {
		if len(group.memberPositions) < 2 {
			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d must contain at least two member positions", groupIndex)
		}
		canonicalMember := false
		for memberIndex, position := range group.memberPositions {
			if position < 0 || position >= recordCount {
				return nil, fmt.Errorf("apply semantic reconciliation plan: group %d member position %d is outside record range [0,%d)", groupIndex, position, recordCount)
			}
			if memberIndex > 0 && group.memberPositions[memberIndex-1] >= position {
				return nil, fmt.Errorf("apply semantic reconciliation plan: group %d member positions must be ascending and unique", groupIndex)
			}
			if _, exists := used[position]; exists {
				return nil, fmt.Errorf("apply semantic reconciliation plan: record position %d belongs to multiple groups", position)
			}
			used[position] = struct{}{}
			canonicalMember = canonicalMember || position == group.canonicalPosition
		}
		if group.canonicalPosition < 0 || group.canonicalPosition >= recordCount {
			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d canonical position %d is outside record range [0,%d)", groupIndex, group.canonicalPosition, recordCount)
		}
		if !canonicalMember {
			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d canonical position %d is not a member", groupIndex, group.canonicalPosition)
		}
		groupByFirstMember[group.memberPositions[0]] = group
	}
	return groupByFirstMember, nil
}

func groupProvenance[T any](group PlanGroup, records []Record[T]) GroupProvenance {
	provenance := GroupProvenance{
		memberPositions:   cloneSlice(group.memberPositions),
		canonicalPosition: group.canonicalPosition,
		earliestPosition:  records[group.memberPositions[0]].earliestInputPosition,
	}
	for _, position := range group.memberPositions {
		provenance.originalInputIndexes = append(provenance.originalInputIndexes, records[position].originalInputIndexes...)
		if records[position].earliestInputPosition < provenance.earliestPosition {
			provenance.earliestPosition = records[position].earliestInputPosition
		}
	}
	provenance.originalInputIndexes, _ = normalizeInputIndexes(provenance.originalInputIndexes)
	return provenance
}

func policyInputs[T any](group PlanGroup, records []Record[T], cloneValue CloneValueFunc[T]) ([]T, T) {
	members := make([]T, len(group.memberPositions))
	for index, position := range group.memberPositions {
		members[index] = cloneValue(records[position].value)
	}
	return members, cloneValue(records[group.canonicalPosition].value)
}

func normalizeInputIndexes(indexes []int) ([]int, error) {
	if indexes == nil {
		return nil, nil
	}
	normalized := append([]int{}, indexes...)
	for _, index := range normalized {
		if index < 0 {
			return nil, fmt.Errorf("original input index must not be negative")
		}
	}
	sort.Ints(normalized)
	write := 0
	for _, index := range normalized {
		if write > 0 && normalized[write-1] == index {
			continue
		}
		normalized[write] = index
		write++
	}
	return normalized[:write], nil
}

func cloneRecord[T any](record Record[T]) Record[T] {
	return cloneRecordWith(record, record.cloneValue)
}

func cloneRecordWith[T any](record Record[T], cloneValue CloneValueFunc[T]) Record[T] {
	return Record[T]{
		value:                 cloneValue(record.value),
		originalInputIndexes:  cloneSlice(record.originalInputIndexes),
		earliestInputPosition: record.earliestInputPosition,
		cloneValue:            cloneValue,
	}
}

func cloneRecords[T any](records []Record[T]) []Record[T] {
	if records == nil {
		return nil
	}
	cloned := make([]Record[T], len(records))
	for index, record := range records {
		cloned[index] = cloneRecord(record)
	}
	return cloned
}

func cloneGroupProvenance(provenance GroupProvenance) GroupProvenance {
	return GroupProvenance{
		memberPositions:      cloneSlice(provenance.memberPositions),
		canonicalPosition:    provenance.canonicalPosition,
		originalInputIndexes: cloneSlice(provenance.originalInputIndexes),
		earliestPosition:     provenance.earliestPosition,
	}
}

func cloneSlice[T any](values []T) []T {
	if values == nil {
		return nil
	}
	return append([]T{}, values...)
}

func preserveEmptySlice[S ~[]E, E any](source S, cloned []E) []E {
	if source == nil {
		return nil
	}
	return cloned
}