eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adopt registry-backed item occurrences

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-05 19:55:34 +00:00
parent f91237e9c0
commit 3dfefd0e14
46 changed files with 996 additions and 1034 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
internal/modules/dnd/normalize/itemevents/normalizer.go
View File

@@ -11,6 +11,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
itemeventmodel "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/itemevents"
itemregistry "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/items/registry"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
)
@@ -20,85 +21,126 @@ const (
normalizationPolicy = "dnd.item_events.normalize.v1"
NormalizationPolicy = normalizationPolicy
ReasonCodeDisplayNormalized = "item_event_display_normalized"
ReasonCodeNameCanonicalized = "item_occurrence_name_canonicalized"
ReasonCodeUnknownItemID = "item_occurrence_unknown_item_id"
ReasonCodeSourceRefsNormalized = "source_references_normalized"
ReasonCodeEventsReordered = "item_events_reordered"
ReasonCodeDuplicateCollapsed = "duplicate_item_event_collapsed"
ReasonCodeWarningsOmitted = "item_event_normalization_warnings_omitted"
ReasonCodeOccurrencesReordered = "item_occurrences_reordered"
ReasonCodeDuplicateCollapsed = "duplicate_item_occurrence_collapsed"
ReasonCodeWarningsOmitted = "item_occurrence_normalization_warnings_omitted"
)
const (
ItemRegistryReferenceSlot = itemregistry.ReferenceSlot
ItemRegistryMaxBytes = itemregistry.MaxBytes
)
var requiredCapabilities = []string{"merged"}
var providedCapabilities = []string{"normalized"}
var _ contracts.Normalizer[dnd.ItemEventList] = (*Normalizer)(nil)
var _ contracts.Normalizer[dnd.ItemOccurrenceList] = (*Normalizer)(nil)
var _ contracts.ManifestMetadataProvider = (*Normalizer)(nil)
var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil)
// Options deliberately has no fields: item-event normalization has no
// generated-reference dependency or configurable semantic behavior.
type Options struct{}
type Normalizer struct{}
type Normalizer struct{ itemResolver *itemregistry.Resolver }
func New(Options) *Normalizer { return &Normalizer{} }
func New(_ Options, references ...contracts.ReferenceSet) (*Normalizer, error) {
if len(references) > 1 {
return nil, normalizerErrorf("at most one reference set may be supplied")
}
var referenceSet contracts.ReferenceSet
if len(references) == 1 {
referenceSet = references[0]
}
resolver, err := itemregistry.NewResolver(referenceSet)
if err != nil {
return nil, normalizerErrorf("prepare item registry: %w", err)
}
return &Normalizer{itemResolver: resolver}, nil
}
func (n *Normalizer) Key() string { return Key }
func (n *Normalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (n *Normalizer) ReferenceSlots() []contracts.ReferenceSlot { return referenceSlots() }
func (n *Normalizer) ManifestMetadata() map[string]any {
if n == nil {
if n == nil || n.itemResolver == nil {
return nil
}
return map[string]any{"normalization_policy": normalizationPolicy}
metadata := map[string]any{"normalization_policy": normalizationPolicy}
seeded := n.itemResolver.Seeded()
if seeded.Bound() {
metadata["item_registry_digest"] = seeded.Digest()
metadata["item_count"] = seeded.Count()
}
return metadata
}
func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint {
if n == nil {
if n == nil || n.itemResolver == nil {
return nil
}
return []pipeline.CheckpointFingerprint{{Name: "normalization_policy", Value: normalizationPolicy}}
return []pipeline.CheckpointFingerprint{
{Name: "normalization_policy", Value: normalizationPolicy},
{Name: "item_registry", Value: n.itemResolver.Seeded().ProjectionDigest()},
}
}
func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.ItemEventList]) (contracts.TypedNormalizeResult[dnd.ItemEventList], error) {
if n == nil {
return contracts.TypedNormalizeResult[dnd.ItemEventList]{}, normalizerErrorf("normalizer must not be nil")
func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.ItemOccurrenceList]) (contracts.TypedNormalizeResult[dnd.ItemOccurrenceList], error) {
if n == nil || n.itemResolver == nil {
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{}, normalizerErrorf("normalizer must not be nil")
}
if ctx == nil {
return contracts.TypedNormalizeResult[dnd.ItemEventList]{}, normalizerErrorf("context must not be nil")
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{}, normalizerErrorf("context must not be nil")
}
if err := ctx.Err(); err != nil {
return contracts.TypedNormalizeResult[dnd.ItemEventList]{}, normalizerErrorf("context error before normalize: %w", err)
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{}, normalizerErrorf("context error before normalize: %w", err)
}
registry, err := n.itemResolver.Resolve(req.References)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{}, normalizerErrorf("resolve item registry: %w", err)
}
if !registry.Bound() {
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{}, normalizerErrorf("item registry reference is required")
}
index := source.NewDocumentIndex(req.Source)
value, warnings := normalizeList(req.MergeOutput.Value, index, shared.NewSourceRefOrderFromIndex(index))
return contracts.TypedNormalizeResult[dnd.ItemEventList]{Value: value, Warnings: warnings}, nil
value, warnings := normalizeList(req.MergeOutput.Value, index, shared.NewSourceRefOrderFromIndex(index), registry)
return contracts.TypedNormalizeResult[dnd.ItemOccurrenceList]{Value: value, Warnings: warnings}, nil
}
type normalizedRecord struct {
event dnd.ItemEvent
event dnd.ItemOccurrence
inputIndex int
}
func normalizeList(input dnd.ItemEventList, index source.DocumentIndex, order shared.SourceRefOrder) (dnd.ItemEventList, []contracts.Warning) {
if input.Events == nil {
return dnd.ItemEventList{}, nil
func normalizeList(input dnd.ItemOccurrenceList, index source.DocumentIndex, order shared.SourceRefOrder, registry *itemregistry.Registry) (dnd.ItemOccurrenceList, []contracts.Warning) {
if input.Occurrences == nil {
return dnd.ItemOccurrenceList{}, nil
}
records := make([]normalizedRecord, len(input.Events))
records := make([]normalizedRecord, len(input.Occurrences))
warnings := make([]contracts.Warning, 0)
for index, inputEvent := range input.Events {
event, changedFields, refsChanged := normalizeEvent(inputEvent, order)
for index, inputEvent := range input.Occurrences {
event, changedFields, found, refsChanged := normalizeEvent(inputEvent, order, registry)
records[index] = normalizedRecord{event: event, inputIndex: index}
if len(changedFields) != 0 {
warnings = append(warnings, contracts.Warning{
Scope: eventScope(index),
ReasonCode: ReasonCodeDisplayNormalized,
ReasonCode: ReasonCodeNameCanonicalized,
Message: fmt.Sprintf("input index %d: normalized display whitespace in %s", index,
diagnostics.Aggregate("fields", changedFields)),
})
}
if found && inputEvent.Name != event.Name {
warnings = append(warnings, contracts.Warning{Scope: eventScope(index), ReasonCode: ReasonCodeNameCanonicalized,
Message: fmt.Sprintf("input index %d: item name canonicalized from %s to %s", index, diagnostics.Quote(inputEvent.Name), diagnostics.Quote(event.Name))})
}
if !found {
warnings = append(warnings, contracts.Warning{Scope: eventScope(index), ReasonCode: ReasonCodeUnknownItemID,
Message: fmt.Sprintf("input index %d: item ID %s is not in the supplied registry", index, diagnostics.Quote(inputEvent.ItemID))})
}
if refsChanged {
warnings = append(warnings, contracts.Warning{
Scope: eventScope(index),
@@ -118,24 +160,27 @@ func normalizeList(input dnd.ItemEventList, index source.DocumentIndex, order sh
}
warnings = append(warnings, contracts.Warning{
Scope: eventScope(record.inputIndex),
ReasonCode: ReasonCodeEventsReordered,
ReasonCode: ReasonCodeOccurrencesReordered,
Message: fmt.Sprintf("input index %d moved to normalized position %d", record.inputIndex, position),
})
}
output, duplicateWarnings := collapseDuplicates(records, index, order)
warnings = append(warnings, duplicateWarnings...)
return dnd.ItemEventList{Events: output}, diagnostics.LimitWarnings(warnings, "item_events", ReasonCodeWarningsOmitted)
return dnd.ItemOccurrenceList{Occurrences: output}, diagnostics.LimitWarnings(warnings, "item_occurrences", ReasonCodeWarningsOmitted)
}
func normalizeEvent(input dnd.ItemEvent, order shared.SourceRefOrder) (dnd.ItemEvent, []string, bool) {
func normalizeEvent(input dnd.ItemOccurrence, order shared.SourceRefOrder, registry *itemregistry.Registry) (dnd.ItemOccurrence, []string, bool, bool) {
output := cloneEvent(input)
changedFields := make([]string, 0, 3)
canonical, found := registry.LookupID(input.ItemID)
if found {
output.Name = canonical.Name
}
changedFields := make([]string, 0, 2)
for _, field := range []struct {
name string
value *string
}{
{name: "name", value: &output.Name},
{name: "from", value: &output.From},
{name: "to", value: &output.To},
} {
@@ -146,10 +191,10 @@ func normalizeEvent(input dnd.ItemEvent, order shared.SourceRefOrder) (dnd.ItemE
}
}
output.SourceRefs = order.Canonicalize(input.SourceRefs)
return output, changedFields, !itemeventmodel.SourceRefsEqual(input.SourceRefs, output.SourceRefs)
return output, changedFields, found, !itemeventmodel.SourceRefsEqual(input.SourceRefs, output.SourceRefs)
}
func cloneEvent(input dnd.ItemEvent) dnd.ItemEvent {
func cloneEvent(input dnd.ItemOccurrence) dnd.ItemOccurrence {
output := input
if input.Quantity != nil {
quantity := *input.Quantity
@@ -166,9 +211,9 @@ type duplicateGroup struct {
removed []int
}
func collapseDuplicates(records []normalizedRecord, index source.DocumentIndex, order shared.SourceRefOrder) ([]dnd.ItemEvent, []contracts.Warning) {
func collapseDuplicates(records []normalizedRecord, index source.DocumentIndex, order shared.SourceRefOrder) ([]dnd.ItemOccurrence, []contracts.Warning) {
if len(records) == 0 {
return make([]dnd.ItemEvent, 0), nil
return make([]dnd.ItemOccurrence, 0), nil
}
keep := make([]bool, len(records))
@@ -190,7 +235,7 @@ func collapseDuplicates(records []normalizedRecord, index source.DocumentIndex,
groups[groupIndex].removed = append(groups[groupIndex].removed, record.inputIndex)
}
output := make([]dnd.ItemEvent, 0, len(records))
output := make([]dnd.ItemOccurrence, 0, len(records))
for recordIndex, record := range records {
if keep[recordIndex] {
output = append(output, cloneEvent(record.event))
@@ -214,11 +259,11 @@ func duplicateWarning(retainedIndex int, removed []int) contracts.Warning {
Scope: eventScope(retainedIndex),
ReasonCode: ReasonCodeDuplicateCollapsed,
Message: diagnostics.Aggregate(
fmt.Sprintf("duplicate item event collapsed; retained input index %d", retainedIndex), issues),
fmt.Sprintf("duplicate item occurrence collapsed; retained input index %d", retainedIndex), issues),
}
}
func eventScope(index int) string { return fmt.Sprintf("events[%d]", index) }
func eventScope(index int) string { return fmt.Sprintf("occurrences[%d]", index) }
func ModuleSpec() pipeline.ModuleSpec {
return pipeline.ModuleSpec{
@@ -227,20 +272,32 @@ func ModuleSpec() pipeline.ModuleSpec {
ExecutionClass: contracts.ExecutionClassDeterministic,
Requires: append([]string(nil), requiredCapabilities...),
Provides: append([]string(nil), providedCapabilities...),
ArtifactKind: dnd.ItemEventListKind,
ArtifactKind: dnd.ItemOccurrenceListKind,
ReferenceSlots: referenceSlots(),
}
}
func Register(registry *pipeline.NormalizerRegistry) error {
return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.ItemEventList], error) {
return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.ItemOccurrenceList], error) {
options, err := DecodeOptions(request.Options)
if err != nil {
return nil, err
}
return New(options), nil
return New(options, request.References)
})
}
func referenceSlots() []contracts.ReferenceSlot {
return []contracts.ReferenceSlot{{
Name: ItemRegistryReferenceSlot,
Description: "Required normalized item registry used only for item identity grounding.",
Required: true,
AcceptedMediaTypes: []string{"application/json"},
AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.ItemRegistryKind},
MaxBytes: ItemRegistryMaxBytes,
}}
}
func DecodeOptions(options map[string]any) (Options, error) {
if err := pipeline.RejectUnknownOptions(options); err != nil {
return Options{}, normalizerErrorf("%w", err)
@@ -251,5 +308,5 @@ func DecodeOptions(options map[string]any) (Options, error) {
func validateOptions(options map[string]any) error { _, err := DecodeOptions(options); return err }
func normalizerErrorf(format string, args ...any) error {
return fmt.Errorf("dnd item events normalizer: "+format, args...)
return fmt.Errorf("dnd item occurrences normalizer: "+format, args...)
}