Add multi-source ingest and source registry

Introduce multi-source ingest support: add SourceConfig and SourceRegistry (in-memory) with YAML loader functions; extend channel YAML to include "sources". Stamp SourceID on HL7Message and RoutedPayload, update MLLP server to accept a sourceID and launch one listener per enabled source in main (with backward-compatible single-listener fallback). Add REST endpoints for CRUD of sources and wire the source registry into the HTTP server. Add channel.SourceIDs filter and router logic to skip channels not matching the message SourceID. Update tests and docs/PROJECT-SCOPE to reflect Phase 4.5 multi-source work; use sync.WaitGroup for concurrent MLLP listeners and ensure graceful handling.

Author: vagnercazarotto

cmd/gateway/main.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 	"time"
 
@@ -45,10 +46,15 @@ func main() {
 
 	// Channel registry — load from YAML if configured.
 	reg := channel.NewRegistry()
+	// Source registry — load from YAML if configured.
+	sourceReg := channel.NewSourceRegistry()
 	if cfg.ChannelsFile != "" {
 		if err := channel.LoadFile(cfg.ChannelsFile, reg); err != nil {
 			log.Printf("[gateway] channels: %v (continuing with empty registry)", err)
 		}
+		if err := channel.LoadSourcesFile(cfg.ChannelsFile, sourceReg); err != nil {
+			log.Printf("[gateway] sources: %v (continuing with empty source registry)", err)
+		}
 	}
 
 	// Dead-letter writer for delivery failures (one shared dir for the gateway).
@@ -64,7 +70,7 @@ func main() {
 	// HTTP REST API
 	httpSrv := &http.Server{
 		Addr:    ":" + cfg.HTTPPort,
-		Handler: rest.New(st, reg).WithAuditDir(cfg.AuditDir),
+		Handler: rest.New(st, reg, sourceReg).WithAuditDir(cfg.AuditDir),
 	}
 	go func() {
 		log.Printf("[gateway] REST API listening on :%s", cfg.HTTPPort)
@@ -73,12 +79,44 @@ func main() {
 		}
 	}()
 
-	// MLLP listener (blocks until ctx cancelled)
-	mllpSrv := mllp.New(cfg.MLLPAddr, makePipeline(rtr, st))
-	if err := mllpSrv.ListenAndServe(ctx); err != nil {
-		log.Printf("[gateway] mllp: %v", err)
+	// Launch one MLLP listener goroutine per enabled source defined in YAML.
+	// If no sources are defined, fall back to the single legacy MLLPAddr.
+	sources := sourceReg.List()
+	activeSources := make([]channel.SourceConfig, 0, len(sources))
+	for _, src := range sources {
+		if src.Enabled && src.Type == channel.SourceMLLP {
+			activeSources = append(activeSources, src)
+		}
+	}
+
+	if len(activeSources) == 0 {
+		// Backward-compat: no sources configured in YAML → use the env-var address.
+		activeSources = []channel.SourceConfig{{
+			ID:      "default",
+			Name:    "Default MLLP Listener",
+			Type:    channel.SourceMLLP,
+			Addr:    cfg.MLLPAddr,
+			Enabled: true,
+		}}
 	}
 
+	var wg sync.WaitGroup
+	for _, src := range activeSources {
+		src := src // capture loop variable
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			srv := mllp.New(src.Addr, src.ID, makePipeline(rtr, st))
+			log.Printf("[gateway] mllp source=%s (%s) listening on %s", src.ID, src.Name, src.Addr)
+			if err := srv.ListenAndServe(ctx); err != nil {
+				log.Printf("[gateway] mllp source=%s: %v", src.ID, err)
+			}
+		}()
+	}
+
+	// Block until all MLLP listeners have stopped (ctx cancelled → listeners exit).
+	wg.Wait()
+
 	// Graceful HTTP shutdown
 	shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -149,7 +187,7 @@ func makePipeline(rtr *router.Router, st store.Store) func(model.HL7Message) err
 			Findings:     len(report.Findings),
 		})
 
-		payload := model.RoutedPayload{Resource: resource, Quality: report, RawHL7: msg.Payload}
+		payload := model.RoutedPayload{Resource: resource, Quality: report, RawHL7: msg.Payload, SourceID: msg.SourceID}
 
 		// Persist as pending before delivery so the message is visible in
 		// the UI even if routing crashes or the gateway restarts mid-flight.

docs/PROJECT-SCOPE.md

@@ -124,14 +124,59 @@ services:
 
 ---
 
-### Phase 4 — Multi-Message Type Support
-
-- ORM^O01 (order messages)
-- ORU^R01 (lab results / observations)
-- SIU^S12 (scheduling)
-- MDM^T02 (clinical documents)
-- Pluggable mapping rules per message type
-- Extended quality rules per message type
+### Phase 4 — Multi-Message Type Support ✅ Complete
+
+- [x] ORM^O01 (order messages) → FHIR R4 ServiceRequest
+- [x] ORU^R01 (lab results / observations) → FHIR R4 DiagnosticReport + Observation
+- [x] SIU^S12 (scheduling) → FHIR R4 Appointment
+- [x] MDM^T02 (clinical documents) → FHIR R4 DocumentReference
+- [x] Pluggable mapping rules per message type (dispatcher by MSH-9.1)
+- [x] Extended quality rules per message type
+
+**Also completed (outside original scope):**
+- [x] HL7 passthrough output — channels can deliver raw HL7v2 via MLLP TCP instead of FHIR JSON
+  - New `output_type: fhir | hl7_passthrough` field on Channel (YAML + REST API + UI)
+  - MLLP client destination adapter with VT/FS/CR framing, ACK/NACK detection, timeout
+  - Fanout already works: one message → FHIR channel + HL7 passthrough channel simultaneously
+- [x] UI improvements: SVG icons, skeleton loading, auto-refresh Dashboard, Toast notifications,
+  Channel confirm-delete modal, Message search, Output type badge in Channels table
+
+---
+
+### Phase 4.5 — Multi-Source / Multi-Pipeline (In Progress)
+
+> **Goal:** a single gateway instance can listen on multiple MLLP ports and/or HTTP endpoints,
+> each bound to its own set of destination channels. Messages from source A only go to channels
+> configured for that source.
+
+```
+[MLLP :2575  (EHR A)] ──→ Pipeline 1 ──→ [FHIR channel A1] [HL7 channel A2]
+[MLLP :2576  (EHR B)] ──→ Pipeline 2 ──→ [FHIR channel B1]
+[HTTP /ingest (API )] ──→ Pipeline 3 ──→ [FHIR channel C1] [HL7 channel C2]
+```
+
+**Implementation roadmap:**
+
+#### Phase A — Multiple MLLP Sources (current)
+- [ ] `SourceConfig` struct: `{ id, type: mllp|http|file, addr, tls_cert, tls_key }`
+- [ ] `sources:` list in channels YAML / REST API
+- [ ] `main.go` launches one `mllp.Server` goroutine per enabled source
+- [ ] `HL7Message.SourceID` carries which source received the message
+- [ ] Router filters channels by `SourceID` when set (empty = accept all)
+
+**Success criteria:** two simultaneous MLLP listeners on different ports, each delivering
+to independent channel sets, verified by integration test.
+
+#### Phase B — Pipeline Model
+- [ ] `Pipeline` struct: `{ id, name, source_id, filters{event_types, min_score}, destination_ids[] }`
+- [ ] Pipeline Registry (CRUD via REST API + YAML)
+- [ ] Router dispatches per pipeline instead of global channel list
+- [ ] UI: Pipeline Manager page (source + filters + destinations)
+
+#### Phase C — HTTP Ingest Source
+- [ ] `POST /api/ingest/hl7` — accepts raw HL7v2 text body, feeds into pipeline
+- [ ] `POST /api/ingest/fhir` — accepts FHIR JSON, validates, routes to channels
+- [ ] Source authentication via API key (Phase 5 prerequisite)
 
 ---
 

internal/api/rest/server.go

@@ -9,6 +9,11 @@
 //	GET /api/v1/channels/{id}     — get channel by ID
 //	PUT /api/v1/channels/{id}     — update channel
 //	DELETE /api/v1/channels/{id}  — delete channel
+//	GET /api/v1/sources           — list sources
+//	POST /api/v1/sources          — create source
+//	GET /api/v1/sources/{id}      — get source by ID
+//	PUT /api/v1/sources/{id}      — update source
+//	DELETE /api/v1/sources/{id}   — delete source
 //	GET /healthz                  — liveness probe
 //	GET /readyz                   — readiness probe (checks store)
 //
@@ -58,13 +63,14 @@ type MessageResponse struct {
 type Server struct {
 	store    store.Store
 	channels *channel.Registry
+	sources  *channel.SourceRegistry
 	auditDir string
 	mux      *http.ServeMux
 }
 
-// New creates a Server backed by st and reg, and registers all routes.
-func New(st store.Store, reg *channel.Registry) *Server {
-	s := &Server{store: st, channels: reg}
+// New creates a Server backed by st, reg and sourceReg, and registers all routes.
+func New(st store.Store, reg *channel.Registry, sourceReg *channel.SourceRegistry) *Server {
+	s := &Server{store: st, channels: reg, sources: sourceReg}
 	s.mux = http.NewServeMux()
 	// message routes
 	s.mux.HandleFunc("GET /api/v1/messages", s.listMessages)
@@ -75,6 +81,12 @@ func New(st store.Store, reg *channel.Registry) *Server {
 	s.mux.HandleFunc("GET /api/v1/channels/{id}", s.getChannel)
 	s.mux.HandleFunc("PUT /api/v1/channels/{id}", s.updateChannel)
 	s.mux.HandleFunc("DELETE /api/v1/channels/{id}", s.deleteChannel)
+	// source routes
+	s.mux.HandleFunc("GET /api/v1/sources", s.listSources)
+	s.mux.HandleFunc("POST /api/v1/sources", s.createSource)
+	s.mux.HandleFunc("GET /api/v1/sources/{id}", s.getSource)
+	s.mux.HandleFunc("PUT /api/v1/sources/{id}", s.updateSource)
+	s.mux.HandleFunc("DELETE /api/v1/sources/{id}", s.deleteSource)
 	// audit + reports routes
 	s.mux.HandleFunc("GET /api/v1/audit", s.listAudit)
 	s.mux.HandleFunc("GET /api/v1/reports", s.getReports)
@@ -258,6 +270,94 @@ func (s *Server) deleteChannel(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
+// ---- source handlers -------------------------------------------------------
+
+// listSources handles GET /api/v1/sources
+func (s *Server) listSources(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, s.sources.List())
+}
+
+// createSource handles POST /api/v1/sources
+func (s *Server) createSource(w http.ResponseWriter, r *http.Request) {
+	var src channel.SourceConfig
+	if err := json.NewDecoder(r.Body).Decode(&src); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	if src.ID == "" {
+		writeError(w, http.StatusBadRequest, "id is required")
+		return
+	}
+	if src.Addr == "" {
+		writeError(w, http.StatusBadRequest, "addr is required")
+		return
+	}
+	if err := s.sources.Add(src); err != nil {
+		if errors.Is(err, channel.ErrDuplicateID) {
+			writeError(w, http.StatusConflict, "source ID already exists")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to create source")
+		return
+	}
+	got, _ := s.sources.Get(src.ID)
+	writeJSON(w, http.StatusCreated, got)
+}
+
+// getSource handles GET /api/v1/sources/{id}
+func (s *Server) getSource(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	src, err := s.sources.Get(id)
+	if err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "source not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to retrieve source")
+		return
+	}
+	writeJSON(w, http.StatusOK, src)
+}
+
+// updateSource handles PUT /api/v1/sources/{id}
+func (s *Server) updateSource(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	var src channel.SourceConfig
+	if err := json.NewDecoder(r.Body).Decode(&src); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	src.ID = id // path wins over body
+	if src.Addr == "" {
+		writeError(w, http.StatusBadRequest, "addr is required")
+		return
+	}
+	if err := s.sources.Update(src); err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "source not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to update source")
+		return
+	}
+	got, _ := s.sources.Get(id)
+	writeJSON(w, http.StatusOK, got)
+}
+
+// deleteSource handles DELETE /api/v1/sources/{id}
+func (s *Server) deleteSource(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	if err := s.sources.Delete(id); err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "source not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to delete source")
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
 // listAudit handles GET /api/v1/audit
 //
 // Query parameters:

internal/api/rest/server_test.go

@@ -89,12 +89,12 @@ func (m *mockStore) Close() error { return nil }
 
 // newSrv builds a Server with the given store and an empty channel registry.
 func newSrv(st store.Store) *rest.Server {
-	return rest.New(st, channel.NewRegistry())
+	return rest.New(st, channel.NewRegistry(), channel.NewSourceRegistry())
 }
 
 // newSrvWithReg builds a Server with the given store and registry.
 func newSrvWithReg(st store.Store, reg *channel.Registry) *rest.Server {
-	return rest.New(st, reg)
+	return rest.New(st, reg, channel.NewSourceRegistry())
 }
 
 func get(srv http.Handler, path string) *httptest.ResponseRecorder {