Remove legacy raw pipeline contracts
This commit is contained in:
@@ -1,234 +1,15 @@
|
||||
package appendorder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"mime"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
||||
)
|
||||
|
||||
const Key = "appendorder"
|
||||
|
||||
var _ contracts.LegacyRawMerger = (*Merger)(nil)
|
||||
|
||||
type Merger struct{}
|
||||
|
||||
func New() *Merger {
|
||||
return &Merger{}
|
||||
}
|
||||
|
||||
func (m *Merger) Key() string {
|
||||
return Key
|
||||
}
|
||||
|
||||
func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
|
||||
if m == nil {
|
||||
return contracts.MergeResult{}, mergerErrorf("merger must not be nil")
|
||||
}
|
||||
if ctx == nil {
|
||||
return contracts.MergeResult{}, mergerErrorf("context must not be nil")
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return contracts.MergeResult{}, mergerErrorf("context error before merge: %w", err)
|
||||
}
|
||||
|
||||
outputs, err := orderedOutputs(req.ExtractOutputs)
|
||||
if err != nil {
|
||||
return contracts.MergeResult{}, err
|
||||
}
|
||||
if len(outputs) == 1 {
|
||||
payload := cloneRawPayload(outputs[0].Payload)
|
||||
return contracts.MergeResult{
|
||||
Output: contracts.MergeOutput{
|
||||
LaneID: req.LaneID,
|
||||
MergerKey: Key,
|
||||
SourceID: outputs[0].SourceID,
|
||||
Schema: outputs[0].Schema,
|
||||
Payload: payload,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
content, err := mergedContent(outputs)
|
||||
if err != nil {
|
||||
return contracts.MergeResult{}, err
|
||||
}
|
||||
return contracts.MergeResult{
|
||||
Output: contracts.MergeOutput{
|
||||
LaneID: req.LaneID,
|
||||
MergerKey: Key,
|
||||
SourceID: sourceID(outputs),
|
||||
Schema: commonSchema(outputs),
|
||||
Payload: contracts.RawPayload{
|
||||
Content: content,
|
||||
MediaType: "application/json",
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func ModuleSpec() pipeline.ModuleSpec {
|
||||
return pipeline.ModuleSpec{
|
||||
Key: Key,
|
||||
Stage: pipeline.StageMerge,
|
||||
Provides: []string{"merged"},
|
||||
}
|
||||
}
|
||||
|
||||
func Register(registry *pipeline.MergerRegistry) error {
|
||||
return registry.RegisterLegacyRawWithSpec(ModuleSpec(), func() (contracts.LegacyRawMerger, error) {
|
||||
return New(), nil
|
||||
})
|
||||
}
|
||||
|
||||
func orderedOutputs(outputs []contracts.ExtractOutput) ([]contracts.ExtractOutput, error) {
|
||||
ordered := make([]contracts.ExtractOutput, 0, len(outputs))
|
||||
for _, output := range outputs {
|
||||
if !isJSONMediaType(output.Payload.MediaType) {
|
||||
return nil, mergerErrorf("extract output for chunk %q has unsupported media type %q", output.ChunkID, output.Payload.MediaType)
|
||||
}
|
||||
if !json.Valid(output.Payload.Content) {
|
||||
return nil, mergerErrorf("extract output for chunk %q contains invalid JSON", output.ChunkID)
|
||||
}
|
||||
ordered = append(ordered, cloneExtractOutput(output))
|
||||
}
|
||||
sort.SliceStable(ordered, func(i, j int) bool {
|
||||
return ordered[i].ChunkIndex < ordered[j].ChunkIndex
|
||||
})
|
||||
return ordered, nil
|
||||
}
|
||||
|
||||
func mergedContent(outputs []contracts.ExtractOutput) ([]byte, error) {
|
||||
values := make([]any, 0, len(outputs))
|
||||
objects := make([]map[string]any, 0, len(outputs))
|
||||
for _, output := range outputs {
|
||||
var value any
|
||||
if err := json.Unmarshal(output.Payload.Content, &value); err != nil {
|
||||
return nil, mergerErrorf("decode extract output for chunk %q: %w", output.ChunkID, err)
|
||||
}
|
||||
values = append(values, value)
|
||||
object, ok := value.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
objects = append(objects, object)
|
||||
}
|
||||
|
||||
if len(objects) == len(outputs) {
|
||||
if field, ok := commonArrayField(objects); ok {
|
||||
merged := make([]any, 0)
|
||||
for _, object := range objects {
|
||||
items := object[field].([]any)
|
||||
merged = append(merged, items...)
|
||||
}
|
||||
return marshalMerged(map[string]any{field: merged})
|
||||
}
|
||||
}
|
||||
return marshalMerged(values)
|
||||
}
|
||||
|
||||
func commonArrayField(objects []map[string]any) (string, bool) {
|
||||
if len(objects) == 0 {
|
||||
return "", false
|
||||
}
|
||||
candidates := map[string]struct{}{}
|
||||
for key, value := range objects[0] {
|
||||
if _, ok := value.([]any); ok {
|
||||
candidates[key] = struct{}{}
|
||||
}
|
||||
}
|
||||
for _, object := range objects[1:] {
|
||||
for key := range candidates {
|
||||
if _, ok := object[key].([]any); !ok {
|
||||
delete(candidates, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(candidates) != 1 {
|
||||
return "", false
|
||||
}
|
||||
for key := range candidates {
|
||||
return key, true
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
func marshalMerged(value any) ([]byte, error) {
|
||||
content, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return nil, mergerErrorf("encode merged output: %w", err)
|
||||
}
|
||||
return content, nil
|
||||
}
|
||||
|
||||
func isJSONMediaType(mediaType string) bool {
|
||||
base, _, err := mime.ParseMediaType(strings.TrimSpace(mediaType))
|
||||
if err != nil {
|
||||
base = strings.TrimSpace(mediaType)
|
||||
}
|
||||
return strings.EqualFold(base, "application/json")
|
||||
}
|
||||
|
||||
func sourceID(outputs []contracts.ExtractOutput) string {
|
||||
for _, output := range outputs {
|
||||
if output.SourceID != "" {
|
||||
return output.SourceID
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func commonSchema(outputs []contracts.ExtractOutput) contracts.ResponseSchema {
|
||||
if len(outputs) == 0 {
|
||||
return contracts.ResponseSchema{}
|
||||
}
|
||||
schema := outputs[0].Schema
|
||||
for _, output := range outputs[1:] {
|
||||
if !sameResponseSchema(output.Schema, schema) {
|
||||
return contracts.ResponseSchema{}
|
||||
}
|
||||
}
|
||||
return schema
|
||||
}
|
||||
|
||||
func sameResponseSchema(left contracts.ResponseSchema, right contracts.ResponseSchema) bool {
|
||||
return left.ID == right.ID && left.Name == right.Name && left.Version == right.Version && string(left.JSONSchema) == string(right.JSONSchema)
|
||||
}
|
||||
|
||||
func cloneExtractOutput(output contracts.ExtractOutput) contracts.ExtractOutput {
|
||||
output.Schema = cloneResponseSchema(output.Schema)
|
||||
output.Payload = cloneRawPayload(output.Payload)
|
||||
return output
|
||||
}
|
||||
|
||||
func cloneResponseSchema(schema contracts.ResponseSchema) contracts.ResponseSchema {
|
||||
schema.JSONSchema = append([]byte(nil), schema.JSONSchema...)
|
||||
return schema
|
||||
}
|
||||
|
||||
func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload {
|
||||
return contracts.RawPayload{
|
||||
Content: append([]byte(nil), payload.Content...),
|
||||
MediaType: payload.MediaType,
|
||||
Metadata: cloneMetadata(payload.Metadata),
|
||||
Warnings: append([]contracts.Warning(nil), payload.Warnings...),
|
||||
}
|
||||
}
|
||||
|
||||
func cloneMetadata(metadata map[string]any) map[string]any {
|
||||
if len(metadata) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]any, len(metadata))
|
||||
for key, value := range metadata {
|
||||
out[key] = value
|
||||
}
|
||||
return out
|
||||
return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageMerge, Provides: []string{"merged"}}
|
||||
}
|
||||
|
||||
func mergerErrorf(format string, args ...any) error {
|
||||
|
||||
@@ -1,215 +0,0 @@
|
||||
package appendorder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
||||
)
|
||||
|
||||
func TestModuleSpecAndRegister(t *testing.T) {
|
||||
want := pipeline.ModuleSpec{
|
||||
Key: Key,
|
||||
Stage: pipeline.StageMerge,
|
||||
Provides: []string{"merged"},
|
||||
}
|
||||
if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
|
||||
}
|
||||
|
||||
registry := pipeline.NewMergerRegistry()
|
||||
if err := Register(registry); err != nil {
|
||||
t.Fatalf("Register() error = %v, want nil", err)
|
||||
}
|
||||
spec, ok := registry.Spec(Key)
|
||||
if !ok {
|
||||
t.Fatalf("Spec(%q) ok = false, want true", Key)
|
||||
}
|
||||
if !reflect.DeepEqual(spec, want) {
|
||||
t.Fatalf("registered spec = %#v, want %#v", spec, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergePassesThroughSingleExtractOutput(t *testing.T) {
|
||||
input := extractOutput("chunk-0", 0, `{"name":"original"}`)
|
||||
|
||||
result, err := New().Merge(context.Background(), contracts.MergeRequest{
|
||||
LaneID: "events",
|
||||
ExtractOutputs: []contracts.ExtractOutput{input},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Merge() error = %v, want nil", err)
|
||||
}
|
||||
|
||||
if result.Output.LaneID != "events" || result.Output.MergerKey != Key {
|
||||
t.Fatalf("output provenance = %#v, want lane and merger", result.Output)
|
||||
}
|
||||
if string(result.Output.Payload.Content) != `{"name":"original"}` {
|
||||
t.Fatalf("content = %s, want original content", result.Output.Payload.Content)
|
||||
}
|
||||
if result.Output.Payload.Metadata["name"] != "chunk-0" {
|
||||
t.Fatalf("metadata = %#v, want original metadata", result.Output.Payload.Metadata)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeDefensivelyCopiesRawPayload(t *testing.T) {
|
||||
input := extractOutput("chunk-0", 0, `{"name":"original"}`)
|
||||
|
||||
result, err := New().Merge(context.Background(), contracts.MergeRequest{
|
||||
LaneID: "events",
|
||||
ExtractOutputs: []contracts.ExtractOutput{input},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Merge() error = %v, want nil", err)
|
||||
}
|
||||
|
||||
input.Payload.Content[0] = '['
|
||||
input.Payload.Metadata["name"] = "changed"
|
||||
|
||||
if string(result.Output.Payload.Content) != `{"name":"original"}` {
|
||||
t.Fatalf("content changed after input mutation: %s", result.Output.Payload.Content)
|
||||
}
|
||||
if result.Output.Payload.Metadata["name"] != "chunk-0" {
|
||||
t.Fatalf("metadata changed after input mutation: %#v", result.Output.Payload.Metadata)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeConcatenatesCommonTopLevelArrayFieldInChunkOrder(t *testing.T) {
|
||||
result, err := New().Merge(context.Background(), contracts.MergeRequest{
|
||||
LaneID: "events",
|
||||
ExtractOutputs: []contracts.ExtractOutput{
|
||||
extractOutput("chunk-1", 1, `{"events":[{"name":"second"}]}`),
|
||||
extractOutput("chunk-0", 0, `{"events":[{"name":"first"}]}`),
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Merge() error = %v, want nil", err)
|
||||
}
|
||||
if result.Output.Payload.MediaType != "application/json" {
|
||||
t.Fatalf("MediaType = %q, want application/json", result.Output.Payload.MediaType)
|
||||
}
|
||||
|
||||
var decoded struct {
|
||||
Events []struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"events"`
|
||||
}
|
||||
if err := json.Unmarshal(result.Output.Payload.Content, &decoded); err != nil {
|
||||
t.Fatalf("Unmarshal() error = %v, want nil", err)
|
||||
}
|
||||
if len(decoded.Events) != 2 || decoded.Events[0].Name != "first" || decoded.Events[1].Name != "second" {
|
||||
t.Fatalf("events = %#v, want concatenated chunk order", decoded.Events)
|
||||
}
|
||||
if result.Output.Schema.ID != "schema-id" {
|
||||
t.Fatalf("schema = %#v, want common extract schema", result.Output.Schema)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeFallsBackToOrderedJSONValueArrayWhenShapesDiffer(t *testing.T) {
|
||||
result, err := New().Merge(context.Background(), contracts.MergeRequest{
|
||||
LaneID: "events",
|
||||
ExtractOutputs: []contracts.ExtractOutput{
|
||||
extractOutput("chunk-1", 1, `{"notes":["second"]}`),
|
||||
extractOutput("chunk-0", 0, `{"events":[{"name":"first"}]}`),
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Merge() error = %v, want nil", err)
|
||||
}
|
||||
|
||||
var decoded []map[string]any
|
||||
if err := json.Unmarshal(result.Output.Payload.Content, &decoded); err != nil {
|
||||
t.Fatalf("Unmarshal() error = %v, want nil", err)
|
||||
}
|
||||
if len(decoded) != 2 {
|
||||
t.Fatalf("len(decoded) = %d, want 2", len(decoded))
|
||||
}
|
||||
if _, ok := decoded[0]["events"]; !ok {
|
||||
t.Fatalf("decoded[0] = %#v, want first chunk value", decoded[0])
|
||||
}
|
||||
if _, ok := decoded[1]["notes"]; !ok {
|
||||
t.Fatalf("decoded[1] = %#v, want second chunk value", decoded[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeRejectsInvalidJSONAndNonJSONMediaTypes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
output contracts.ExtractOutput
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "invalid JSON",
|
||||
output: extractOutput("chunk-0", 0, `{"events":[`),
|
||||
want: "invalid JSON",
|
||||
},
|
||||
{
|
||||
name: "non JSON media type",
|
||||
output: func() contracts.ExtractOutput {
|
||||
output := extractOutput("chunk-0", 0, `{"events":[]}`)
|
||||
output.Payload.MediaType = "text/plain"
|
||||
return output
|
||||
}(),
|
||||
want: "unsupported media type",
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
_, err := New().Merge(context.Background(), contracts.MergeRequest{
|
||||
LaneID: "events",
|
||||
ExtractOutputs: []contracts.ExtractOutput{test.output},
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("Merge() error = nil, want error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), test.want) {
|
||||
t.Fatalf("Merge() error = %q, want %q", err.Error(), test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTypedMergeUsesRequestOrderForReusableValueType(t *testing.T) {
|
||||
type notes struct{ Values []string }
|
||||
merger, err := NewTyped(func(values []notes) (notes, error) {
|
||||
var combined notes
|
||||
for _, value := range values {
|
||||
combined.Values = append(combined.Values, value.Values...)
|
||||
}
|
||||
return combined, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewTyped() error = %v", err)
|
||||
}
|
||||
result, err := merger.Merge(context.Background(), contracts.TypedMergeRequest[notes]{ExtractOutputs: []contracts.ExtractArtifact[notes]{
|
||||
{ChunkIndex: 4, Value: notes{Values: []string{"first"}}},
|
||||
{ChunkIndex: 1, Value: notes{Values: []string{"second"}}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatalf("Merge() error = %v", err)
|
||||
}
|
||||
if got := result.Value.Values; !reflect.DeepEqual(got, []string{"first", "second"}) {
|
||||
t.Fatalf("Values = %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func extractOutput(chunkID string, chunkIndex int, content string) contracts.ExtractOutput {
|
||||
return contracts.ExtractOutput{
|
||||
LaneID: "events",
|
||||
ExtractorKey: "extract",
|
||||
SourceID: "source-1",
|
||||
ChunkID: chunkID,
|
||||
ChunkIndex: chunkIndex,
|
||||
Schema: contracts.ResponseSchema{ID: "schema-id", Name: "schema-name", Version: "v1"},
|
||||
Payload: contracts.RawPayload{
|
||||
Content: []byte(content),
|
||||
MediaType: "application/json",
|
||||
Metadata: map[string]any{"name": chunkID},
|
||||
},
|
||||
}
|
||||
}
|
||||
25
internal/modules/generic/merge/appendorder/typed_test.go
Normal file
25
internal/modules/generic/merge/appendorder/typed_test.go
Normal file
@@ -0,0 +1,25 @@
|
||||
package appendorder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
)
|
||||
|
||||
func TestTypedMergerCombinesValuesInFrameworkOrder(t *testing.T) {
|
||||
merger, err := NewTyped(func(values []string) (string, error) {
|
||||
if !reflect.DeepEqual(values, []string{"first", "second"}) {
|
||||
t.Fatalf("values=%#v", values)
|
||||
}
|
||||
return values[0] + values[1], nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := merger.Merge(context.Background(), contracts.TypedMergeRequest[string]{ExtractOutputs: []contracts.ExtractArtifact[string]{{ChunkIndex: 0, Value: "first"}, {ChunkIndex: 1, Value: "second"}}})
|
||||
if err != nil || result.Value != "firstsecond" {
|
||||
t.Fatalf("result=%#v err=%v", result, err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user