Migrate location reconciliation to shared engine
This commit is contained in:
@@ -4,54 +4,66 @@ import (
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/locations/identity"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
|
||||
)
|
||||
|
||||
type safeReconciliationGroup struct {
|
||||
members []int
|
||||
canonical int
|
||||
}
|
||||
|
||||
func reconciliationCandidates(records []normalizedRecord) []entityreconcile.Candidate {
|
||||
candidates := make([]entityreconcile.Candidate, len(records))
|
||||
func reconciliationInputs(records []normalizedRecord) ([]semanticreconcile.Candidate, []semanticreconcile.Record[dnd.Location], error) {
|
||||
candidates := make([]semanticreconcile.Candidate, len(records))
|
||||
envelopes := make([]semanticreconcile.Record[dnd.Location], len(records))
|
||||
for index, record := range records {
|
||||
candidates[index] = entityreconcile.Candidate{Name: record.location.Name, SourceRefs: cloneSourceRefs(record.location.SourceRefs)}
|
||||
candidates[index] = semanticreconcile.Candidate{
|
||||
Label: record.location.Name,
|
||||
SourceRefs: cloneSourceRefs(record.location.SourceRefs),
|
||||
}
|
||||
envelope, err := semanticreconcile.NewRecord(record.location, record.inputIndexes, record.earliest, cloneLocation)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("record %d: %w", index, err)
|
||||
}
|
||||
envelopes[index] = envelope
|
||||
}
|
||||
return candidates
|
||||
return candidates, envelopes, nil
|
||||
}
|
||||
|
||||
func reconciliationGroups(assessment entityreconcile.Assessment, candidateKeys []string) []safeReconciliationGroup {
|
||||
positions := make(map[string]int, len(candidateKeys))
|
||||
for index, key := range candidateKeys {
|
||||
positions[key] = index
|
||||
}
|
||||
safeGroups := assessment.SafeGroups()
|
||||
groups := make([]safeReconciliationGroup, 0, len(safeGroups))
|
||||
for _, group := range safeGroups {
|
||||
members := group.Members()
|
||||
memberPositions := make([]int, len(members))
|
||||
valid := true
|
||||
for index, key := range members {
|
||||
position, ok := positions[key]
|
||||
if !ok {
|
||||
valid = false
|
||||
break
|
||||
func applyReconciliationPlan(plan semanticreconcile.Plan, records []normalizedRecord, envelopes []semanticreconcile.Record[dnd.Location], order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning, error) {
|
||||
application, err := semanticreconcile.ApplyPlan(plan, envelopes, semanticreconcile.ApplicationPolicy[dnd.Location]{
|
||||
CloneValue: cloneLocation,
|
||||
ConsolidateGroup: func(members []dnd.Location, canonical dnd.Location) (dnd.Location, error) {
|
||||
output := cloneLocation(canonical)
|
||||
output.SourceRefs = nil
|
||||
for _, member := range members {
|
||||
output.SourceRefs = append(output.SourceRefs, member.SourceRefs...)
|
||||
}
|
||||
memberPositions[index] = position
|
||||
}
|
||||
canonical, ok := positions[group.Canonical()]
|
||||
if valid && ok {
|
||||
groups = append(groups, safeReconciliationGroup{members: memberPositions, canonical: canonical})
|
||||
output.SourceRefs = order.Canonicalize(output.SourceRefs)
|
||||
output.ID = identity.DeriveID(output.Name, output.SourceRefs)
|
||||
return output, nil
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
applied := application.Records()
|
||||
output := make([]normalizedRecord, len(applied))
|
||||
for index, record := range applied {
|
||||
output[index] = normalizedRecord{
|
||||
location: record.Value(),
|
||||
inputIndexes: record.OriginalInputIndexes(),
|
||||
earliest: record.EarliestInputPosition(),
|
||||
}
|
||||
}
|
||||
return groups
|
||||
warnings := make([]contracts.Warning, 0, len(application.AppliedGroups()))
|
||||
for _, event := range application.AppliedGroups() {
|
||||
provenance := event.Provenance()
|
||||
warnings = append(warnings, semanticDuplicateWarning(provenance, records[provenance.CanonicalPosition()]))
|
||||
}
|
||||
return output, warnings, nil
|
||||
}
|
||||
|
||||
func reconciliationIssues(assessment entityreconcile.Assessment) []string {
|
||||
issues := assessment.Issues()
|
||||
func reconciliationIssues(issues []semanticreconcile.Issue) []string {
|
||||
details := make([]string, len(issues))
|
||||
for index, issue := range issues {
|
||||
details[index] = fmt.Sprintf("group %d: %s", issue.GroupIndex, issue.Category)
|
||||
@@ -59,54 +71,18 @@ func reconciliationIssues(assessment entityreconcile.Assessment) []string {
|
||||
return details
|
||||
}
|
||||
|
||||
func applySafeGroups(records []normalizedRecord, groups []safeReconciliationGroup, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) {
|
||||
byMember := make(map[int]safeReconciliationGroup, len(groups)*2)
|
||||
for _, group := range groups {
|
||||
for _, member := range group.members {
|
||||
byMember[member] = group
|
||||
}
|
||||
}
|
||||
output := make([]normalizedRecord, 0, len(records)-len(groups))
|
||||
warnings := make([]contracts.Warning, 0, len(groups))
|
||||
for index, record := range records {
|
||||
group, grouped := byMember[index]
|
||||
if !grouped {
|
||||
output = append(output, cloneRecord(record))
|
||||
continue
|
||||
}
|
||||
if group.members[0] != index {
|
||||
continue
|
||||
}
|
||||
consolidated := consolidateSemanticGroup(records, group, order)
|
||||
output = append(output, consolidated)
|
||||
warnings = append(warnings, semanticDuplicateWarning(consolidated, records[group.canonical]))
|
||||
}
|
||||
return output, warnings
|
||||
}
|
||||
|
||||
func consolidateSemanticGroup(records []normalizedRecord, group safeReconciliationGroup, order shared.SourceRefOrder) normalizedRecord {
|
||||
output := cloneRecord(records[group.members[0]])
|
||||
output.location.Name = records[group.canonical].location.Name
|
||||
for _, member := range group.members[1:] {
|
||||
output.location.SourceRefs = append(output.location.SourceRefs, records[member].location.SourceRefs...)
|
||||
output.inputIndexes = append(output.inputIndexes, records[member].inputIndexes...)
|
||||
if records[member].earliest < output.earliest {
|
||||
output.earliest = records[member].earliest
|
||||
}
|
||||
}
|
||||
output.inputIndexes = sortedUniqueIndexes(output.inputIndexes)
|
||||
output.location.SourceRefs = order.Canonicalize(output.location.SourceRefs)
|
||||
output.location.ID = identity.DeriveID(output.location.Name, output.location.SourceRefs)
|
||||
return output
|
||||
}
|
||||
|
||||
func semanticDuplicateWarning(record normalizedRecord, canonical normalizedRecord) contracts.Warning {
|
||||
details := make([]string, 0, len(record.inputIndexes)+1)
|
||||
for _, inputIndex := range record.inputIndexes {
|
||||
func semanticDuplicateWarning(provenance semanticreconcile.GroupProvenance, canonical normalizedRecord) contracts.Warning {
|
||||
inputIndexes := provenance.OriginalInputIndexes()
|
||||
details := make([]string, 0, len(inputIndexes)+1)
|
||||
for _, inputIndex := range inputIndexes {
|
||||
details = append(details, fmt.Sprintf("input index %d", inputIndex))
|
||||
}
|
||||
if canonical.earliest != record.earliest {
|
||||
if canonical.earliest != provenance.EarliestInputPosition() {
|
||||
details = append(details, fmt.Sprintf("canonical display name from input index %d", canonical.earliest))
|
||||
}
|
||||
return contracts.Warning{Scope: locationScope(record.earliest), ReasonCode: ReasonCodeDuplicateLocationCollapsed, Message: diagnostics.Aggregate("semantic duplicate consolidation", details)}
|
||||
return contracts.Warning{
|
||||
Scope: locationScope(provenance.EarliestInputPosition()),
|
||||
ReasonCode: ReasonCodeDuplicateLocationCollapsed,
|
||||
Message: diagnostics.Aggregate("semantic duplicate consolidation", details),
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user