395 lines
14 KiB
Go
395 lines
14 KiB
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 warning 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 warning 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
|
|
}
|