Skip to content

Commit df66102

Browse files
vagnercazarottovagnercazarotto
committed
Add channel-aware router scaffold and audit fields
Introduce a registry-aware routing scaffold (Phase 3 PR 1) and wire it into the MLLP pipeline. Changes include: - cmd/gateway: create a router from the channel registry and switch the MLLP listener to use makePipeline(rtr). Extracted the pipeline closure into makePipeline(rtr) which runs parsing, mapping, scoring and delegates routing to the Router. - internal/router: add a Router type, Decision model and a Route method that iterates registry channels, applies simple filters (disabled, MinQualityScore), emits per-channel audit entries and returns decisions. Delivery is stubbed (no HTTP yet); Router handles nil/empty registries gracefully. Added decide helper implementing initial filtering rules. - internal/router: add comprehensive unit tests exercising nil/empty registries, disabled channels, quality thresholds, passing channels and fan-out behavior. - internal/audit: extend audit.Entry with ChannelID and DestURL fields for delivery-stage logging. - web/src/types: expose channel_id and dest_url in the frontend AuditEntry type. Purpose: scaffold channel-based routing and enrich audit logs so future PRs can implement real HTTP delivery, retries and DLQ wiring while preserving observability of routing intent.
1 parent 7d44bdc commit df66102

5 files changed

Lines changed: 280 additions & 70 deletions

File tree

‎cmd/gateway/main.go‎

Lines changed: 70 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,10 @@ func main() {
4949
}
5050
}
5151

52+
// Channel-aware dispatcher (Phase 3 PR 1: scaffold only — real HTTP
53+
// delivery, retry and DLQ wiring follow in subsequent PRs).
54+
rtr := router.New(reg)
55+
5256
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
5357
defer stop()
5458

@@ -65,7 +69,7 @@ func main() {
6569
}()
6670

6771
// MLLP listener (blocks until ctx cancelled)
68-
mllpSrv := mllp.New(cfg.MLLPAddr, pipeline)
72+
mllpSrv := mllp.New(cfg.MLLPAddr, makePipeline(rtr))
6973
if err := mllpSrv.ListenAndServe(ctx); err != nil {
7074
log.Printf("[gateway] mllp: %v", err)
7175
}
@@ -78,69 +82,75 @@ func main() {
7882
fmt.Println("[gateway] shutdown complete")
7983
}
8084

81-
// pipeline runs a received HL7v2 message through the full processing chain,
82-
// emitting one structured audit log line per stage.
83-
func pipeline(msg model.HL7Message) error {
84-
audit.Log(audit.Entry{
85-
MessageID: msg.ID,
86-
Stage: "ingest",
87-
Status: "ok",
88-
})
89-
90-
t := time.Now()
91-
parsed, err := parser.Parse(msg.Payload)
92-
if err != nil {
85+
// makePipeline returns the message-processing closure used by the MLLP
86+
// listener. The closure runs each received HL7v2 message through the full
87+
// processing chain and emits one structured audit log line per stage.
88+
//
89+
// Routing is delegated to rtr, which fans the payload out across the
90+
// channels currently registered (see internal/router).
91+
func makePipeline(rtr *router.Router) func(model.HL7Message) error {
92+
return func(msg model.HL7Message) error {
93+
audit.Log(audit.Entry{
94+
MessageID: msg.ID,
95+
Stage: "ingest",
96+
Status: "ok",
97+
})
98+
99+
t := time.Now()
100+
parsed, err := parser.Parse(msg.Payload)
101+
if err != nil {
102+
audit.Log(audit.Entry{
103+
MessageID: msg.ID,
104+
Stage: "parse",
105+
DurationMs: time.Since(t).Milliseconds(),
106+
Status: "error",
107+
Error: err.Error(),
108+
})
109+
return fmt.Errorf("parse: %w", err)
110+
}
93111
audit.Log(audit.Entry{
94112
MessageID: msg.ID,
95113
Stage: "parse",
96114
DurationMs: time.Since(t).Milliseconds(),
97-
Status: "error",
98-
Error: err.Error(),
115+
Status: "ok",
116+
Segments: len(parsed.Segments),
117+
})
118+
119+
t = time.Now()
120+
resource := mapping.ToFHIR(msg.ID, parsed)
121+
eventType, _ := resource.Body["eventType"].(string)
122+
audit.Log(audit.Entry{
123+
MessageID: msg.ID,
124+
Stage: "map",
125+
DurationMs: time.Since(t).Milliseconds(),
126+
Status: "ok",
127+
ResourceType: resource.ResourceType,
128+
EventType: eventType,
129+
})
130+
131+
t = time.Now()
132+
report := quality.Score(resource)
133+
audit.Log(audit.Entry{
134+
MessageID: msg.ID,
135+
Stage: "score",
136+
DurationMs: time.Since(t).Milliseconds(),
137+
Status: "ok",
138+
Score: audit.F64(report.Score),
139+
Completeness: audit.F64(report.Completeness),
140+
Conformity: audit.F64(report.Conformity),
141+
Confidence: audit.F64(report.Confidence),
142+
Findings: len(report.Findings),
143+
})
144+
145+
t = time.Now()
146+
rtr.Route(context.Background(), model.RoutedPayload{Resource: resource, Quality: report})
147+
audit.Log(audit.Entry{
148+
MessageID: msg.ID,
149+
Stage: "route",
150+
DurationMs: time.Since(t).Milliseconds(),
151+
Status: "ok",
99152
})
100-
return fmt.Errorf("parse: %w", err)
153+
154+
return nil
101155
}
102-
audit.Log(audit.Entry{
103-
MessageID: msg.ID,
104-
Stage: "parse",
105-
DurationMs: time.Since(t).Milliseconds(),
106-
Status: "ok",
107-
Segments: len(parsed.Segments),
108-
})
109-
110-
t = time.Now()
111-
resource := mapping.ToFHIR(msg.ID, parsed)
112-
eventType, _ := resource.Body["eventType"].(string)
113-
audit.Log(audit.Entry{
114-
MessageID: msg.ID,
115-
Stage: "map",
116-
DurationMs: time.Since(t).Milliseconds(),
117-
Status: "ok",
118-
ResourceType: resource.ResourceType,
119-
EventType: eventType,
120-
})
121-
122-
t = time.Now()
123-
report := quality.Score(resource)
124-
audit.Log(audit.Entry{
125-
MessageID: msg.ID,
126-
Stage: "score",
127-
DurationMs: time.Since(t).Milliseconds(),
128-
Status: "ok",
129-
Score: audit.F64(report.Score),
130-
Completeness: audit.F64(report.Completeness),
131-
Conformity: audit.F64(report.Conformity),
132-
Confidence: audit.F64(report.Confidence),
133-
Findings: len(report.Findings),
134-
})
135-
136-
t = time.Now()
137-
router.Route(model.RoutedPayload{Resource: resource, Quality: report})
138-
audit.Log(audit.Entry{
139-
MessageID: msg.ID,
140-
Stage: "route",
141-
DurationMs: time.Since(t).Milliseconds(),
142-
Status: "ok",
143-
})
144-
145-
return nil
146156
}

‎internal/audit/logger.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@ type Entry struct {
3131
Conformity *float64 `json:"conformity,omitempty"`
3232
Confidence *float64 `json:"confidence,omitempty"`
3333
Findings int `json:"findings,omitempty"`
34+
// ChannelID identifies the destination channel for stage="deliver" entries.
35+
ChannelID string `json:"channel_id,omitempty"`
36+
// DestURL is the resolved target URL for the delivery attempt.
37+
DestURL string `json:"dest_url,omitempty"`
3438
}
3539

3640
// Log writes e as a single JSON line to the audit sink.

‎internal/router/router.go‎

Lines changed: 91 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,99 @@
1+
// Package router dispatches a routed payload to every channel registered in
2+
// channel.Registry that satisfies the channel's filters.
3+
//
4+
// PR 1 of Phase 3 introduces the registry-aware scaffold: per-channel decisions
5+
// are produced and audit-logged, but no real HTTP delivery happens yet. Real
6+
// delivery, retry, and dead-lettering wire in via subsequent PRs.
17
package router
28

39
import (
4-
"fmt"
10+
"context"
11+
"log"
512

13+
"github.com/vagnercazarotto/verifhir-gateway/internal/audit"
14+
"github.com/vagnercazarotto/verifhir-gateway/internal/channel"
615
"github.com/vagnercazarotto/verifhir-gateway/internal/model"
716
)
817

9-
// Route dispatches a routed payload to its destination.
10-
// Currently a stdout stub; will be replaced by HTTP / queue adapters in Phase 2.
11-
func Route(payload model.RoutedPayload) {
12-
fmt.Printf("[router] resource=%s id=%s score=%.2f findings=%d\n",
13-
payload.Resource.ResourceType,
14-
payload.Resource.ID,
15-
payload.Quality.Score,
16-
len(payload.Quality.Findings),
17-
)
18+
// Decision is the per-channel routing outcome for one message.
19+
type Decision struct {
20+
ChannelID string
21+
URL string
22+
// Status mirrors the audit "status" value: "pending" (will deliver),
23+
// "skipped" (filter rejected the message), or "no_channels".
24+
Status string
25+
// Reason is human-readable detail for "skipped" decisions.
26+
Reason string
27+
}
28+
29+
// Router holds the registry of delivery channels and produces routing
30+
// decisions for each processed message.
31+
type Router struct {
32+
reg *channel.Registry
33+
}
34+
35+
// New constructs a Router backed by reg. A nil reg is allowed and causes
36+
// Route to log a warning and return an empty decision list (so the pipeline
37+
// can continue processing without channels configured).
38+
func New(reg *channel.Registry) *Router {
39+
return &Router{reg: reg}
40+
}
41+
42+
// Route iterates every channel in the registry, applies its filters, and
43+
// returns a Decision for each. One audit entry with stage="deliver" is emitted
44+
// per decision so operators can trace why a message did or did not reach a
45+
// destination.
46+
//
47+
// Real HTTP delivery is added in PR 2; this implementation only logs intent.
48+
func (r *Router) Route(_ context.Context, payload model.RoutedPayload) []Decision {
49+
if r == nil || r.reg == nil {
50+
log.Printf("[router] no registry wired; skipping msg=%s",
51+
payload.Resource.ID)
52+
return nil
53+
}
54+
55+
channels := r.reg.List()
56+
if len(channels) == 0 {
57+
audit.Log(audit.Entry{
58+
MessageID: payload.Resource.ID,
59+
Stage: "deliver",
60+
Status: "no_channels",
61+
})
62+
return nil
63+
}
64+
65+
decisions := make([]Decision, 0, len(channels))
66+
for _, ch := range channels {
67+
d := decide(ch, payload)
68+
decisions = append(decisions, d)
69+
audit.Log(audit.Entry{
70+
MessageID: payload.Resource.ID,
71+
Stage: "deliver",
72+
Status: d.Status,
73+
ChannelID: d.ChannelID,
74+
DestURL: d.URL,
75+
Error: d.Reason,
76+
})
77+
}
78+
return decisions
79+
}
80+
81+
// decide applies channel filters in priority order:
82+
// 1. disabled channels are skipped
83+
// 2. payloads below MinQualityScore are skipped
84+
//
85+
// Future PRs will add event-type filtering and other rules.
86+
func decide(ch channel.Channel, payload model.RoutedPayload) Decision {
87+
d := Decision{ChannelID: ch.ID, URL: ch.URL}
88+
switch {
89+
case !ch.Enabled:
90+
d.Status = "skipped"
91+
d.Reason = "channel disabled"
92+
case payload.Quality.Score < ch.MinQualityScore:
93+
d.Status = "skipped"
94+
d.Reason = "quality score below channel threshold"
95+
default:
96+
d.Status = "pending"
97+
}
98+
return d
1899
}

‎internal/router/router_test.go‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
package router_test
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
"github.com/vagnercazarotto/verifhir-gateway/internal/channel"
8+
"github.com/vagnercazarotto/verifhir-gateway/internal/model"
9+
"github.com/vagnercazarotto/verifhir-gateway/internal/router"
10+
)
11+
12+
func mustAdd(t *testing.T, reg *channel.Registry, ch channel.Channel) {
13+
t.Helper()
14+
if err := reg.Add(ch); err != nil {
15+
t.Fatalf("add channel %s: %v", ch.ID, err)
16+
}
17+
}
18+
19+
func samplePayload(score float64) model.RoutedPayload {
20+
return model.RoutedPayload{
21+
Resource: model.FHIRResource{ResourceType: "Bundle", ID: "msg-1"},
22+
Quality: model.QualityReport{Score: score},
23+
}
24+
}
25+
26+
func TestRouteWithNilRegistryReturnsNil(t *testing.T) {
27+
r := router.New(nil)
28+
if got := r.Route(context.Background(), samplePayload(0.9)); got != nil {
29+
t.Fatalf("expected nil decisions, got %+v", got)
30+
}
31+
}
32+
33+
func TestRouteWithNoChannelsReturnsNil(t *testing.T) {
34+
r := router.New(channel.NewRegistry())
35+
if got := r.Route(context.Background(), samplePayload(0.9)); got != nil {
36+
t.Fatalf("expected nil decisions for empty registry, got %+v", got)
37+
}
38+
}
39+
40+
func TestRouteSkipsDisabledChannels(t *testing.T) {
41+
reg := channel.NewRegistry()
42+
mustAdd(t, reg, channel.Channel{
43+
ID: "off", URL: "http://example.test", Enabled: false,
44+
})
45+
got := router.New(reg).Route(context.Background(), samplePayload(1.0))
46+
if len(got) != 1 {
47+
t.Fatalf("decisions: got %d want 1", len(got))
48+
}
49+
if got[0].Status != "skipped" || got[0].Reason != "channel disabled" {
50+
t.Fatalf("unexpected decision: %+v", got[0])
51+
}
52+
}
53+
54+
func TestRouteEnforcesQualityThreshold(t *testing.T) {
55+
reg := channel.NewRegistry()
56+
mustAdd(t, reg, channel.Channel{
57+
ID: "strict", URL: "http://example.test",
58+
Enabled: true, MinQualityScore: 0.8,
59+
})
60+
got := router.New(reg).Route(context.Background(), samplePayload(0.5))
61+
if len(got) != 1 || got[0].Status != "skipped" {
62+
t.Fatalf("expected skipped decision, got %+v", got)
63+
}
64+
if got[0].Reason != "quality score below channel threshold" {
65+
t.Fatalf("unexpected reason: %q", got[0].Reason)
66+
}
67+
}
68+
69+
func TestRoutePassesQualifyingMessages(t *testing.T) {
70+
reg := channel.NewRegistry()
71+
mustAdd(t, reg, channel.Channel{
72+
ID: "primary", URL: "http://hapi:8080/fhir",
73+
Enabled: true, MinQualityScore: 0.7,
74+
})
75+
got := router.New(reg).Route(context.Background(), samplePayload(0.9))
76+
if len(got) != 1 || got[0].Status != "pending" {
77+
t.Fatalf("expected pending decision, got %+v", got)
78+
}
79+
if got[0].ChannelID != "primary" || got[0].URL != "http://hapi:8080/fhir" {
80+
t.Fatalf("decision metadata mismatch: %+v", got[0])
81+
}
82+
}
83+
84+
func TestRouteFanOutToMultipleChannels(t *testing.T) {
85+
reg := channel.NewRegistry()
86+
mustAdd(t, reg, channel.Channel{
87+
ID: "a", URL: "http://a", Enabled: true, MinQualityScore: 0.8,
88+
})
89+
mustAdd(t, reg, channel.Channel{
90+
ID: "b", URL: "http://b", Enabled: true, MinQualityScore: 0.8,
91+
})
92+
mustAdd(t, reg, channel.Channel{
93+
ID: "off", URL: "http://off", Enabled: false,
94+
})
95+
96+
got := router.New(reg).Route(context.Background(), samplePayload(0.95))
97+
if len(got) != 3 {
98+
t.Fatalf("decisions: got %d want 3", len(got))
99+
}
100+
101+
pending, skipped := 0, 0
102+
for _, d := range got {
103+
switch d.Status {
104+
case "pending":
105+
pending++
106+
case "skipped":
107+
skipped++
108+
}
109+
}
110+
if pending != 2 || skipped != 1 {
111+
t.Fatalf("unexpected breakdown: pending=%d skipped=%d", pending, skipped)
112+
}
113+
}

‎web/src/types/index.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,8 @@ export interface AuditEntry {
5757
conformity?: number
5858
confidence?: number
5959
findings?: number
60+
channel_id?: string
61+
dest_url?: string
6062
}
6163

6264
// ---- reports ---------------------------------------------------------------

0 commit comments

Comments
 (0)