Add pipelines and sources management

Introduce pipeline support and a sources UI: add Pipeline and PipelineRegistry types with thread-safe CRUD methods in the channel package, wire a pipeline registry into the gateway, and extend the REST server with full pipeline routes (list/create/get/update/delete). Update server constructor and tests to accept the new pipeline registry. On the frontend add Source and Pipeline types, API client methods for sources and pipelines, new pages for Pipelines and Sources (including modals/forms, listing, create/update/delete and toggles), route entries in App, and sidebar icons/links. Enables managing sources and routing pipelines via REST + web UI.

Author: vagnercazarotto

cmd/gateway/main.go

@@ -48,6 +48,8 @@ func main() {
 	reg := channel.NewRegistry()
 	// Source registry — load from YAML if configured.
 	sourceReg := channel.NewSourceRegistry()
+	// Pipeline registry — starts empty; managed via REST API.
+	pipelineReg := channel.NewPipelineRegistry()
 	if cfg.ChannelsFile != "" {
 		if err := channel.LoadFile(cfg.ChannelsFile, reg); err != nil {
 			log.Printf("[gateway] channels: %v (continuing with empty registry)", err)
@@ -70,7 +72,7 @@ func main() {
 	// HTTP REST API
 	httpSrv := &http.Server{
 		Addr:    ":" + cfg.HTTPPort,
-		Handler: rest.New(st, reg, sourceReg).WithAuditDir(cfg.AuditDir),
+		Handler: rest.New(st, reg, sourceReg, pipelineReg).WithAuditDir(cfg.AuditDir),
 	}
 	go func() {
 		log.Printf("[gateway] REST API listening on :%s", cfg.HTTPPort)

internal/api/rest/server.go

@@ -2,20 +2,25 @@
 //
 // Endpoints:
 //
-//	GET /api/v1/messages          — list messages (query: status, limit)
-//	GET /api/v1/messages/{id}     — get a single message by ID
-//	GET /api/v1/channels          — list channels
-//	POST /api/v1/channels         — create channel
-//	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)
+//	GET /api/v1/messages           — list messages (query: status, limit)
+//	GET /api/v1/messages/{id}      — get a single message by ID
+//	GET /api/v1/channels           — list channels
+//	POST /api/v1/channels          — create channel
+//	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 /api/v1/pipelines          — list pipelines
+//	POST /api/v1/pipelines         — create pipeline
+//	GET /api/v1/pipelines/{id}     — get pipeline by ID
+//	PUT /api/v1/pipelines/{id}     — update pipeline
+//	DELETE /api/v1/pipelines/{id}  — delete pipeline
+//	GET /healthz                   — liveness probe
+//	GET /readyz                    — readiness probe (checks store)
 //
 // All responses are JSON. Error bodies have the shape {"error":"reason"}.
 // The server uses Go 1.22 enhanced ServeMux patterns, so unknown methods
@@ -61,16 +66,17 @@ type MessageResponse struct {
 // Server is an http.Handler that exposes the message store and channel
 // registry as a REST API.
 type Server struct {
-	store    store.Store
-	channels *channel.Registry
-	sources  *channel.SourceRegistry
-	auditDir string
-	mux      *http.ServeMux
+	store     store.Store
+	channels  *channel.Registry
+	sources   *channel.SourceRegistry
+	pipelines *channel.PipelineRegistry
+	auditDir  string
+	mux       *http.ServeMux
 }
 
-// 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}
+// New creates a Server backed by st, reg, sourceReg and pipelineReg, and registers all routes.
+func New(st store.Store, reg *channel.Registry, sourceReg *channel.SourceRegistry, pipelineReg *channel.PipelineRegistry) *Server {
+	s := &Server{store: st, channels: reg, sources: sourceReg, pipelines: pipelineReg}
 	s.mux = http.NewServeMux()
 	// message routes
 	s.mux.HandleFunc("GET /api/v1/messages", s.listMessages)
@@ -87,6 +93,12 @@ func New(st store.Store, reg *channel.Registry, sourceReg *channel.SourceRegistr
 	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)
+	// pipeline routes
+	s.mux.HandleFunc("GET /api/v1/pipelines", s.listPipelines)
+	s.mux.HandleFunc("POST /api/v1/pipelines", s.createPipeline)
+	s.mux.HandleFunc("GET /api/v1/pipelines/{id}", s.getPipeline)
+	s.mux.HandleFunc("PUT /api/v1/pipelines/{id}", s.updatePipeline)
+	s.mux.HandleFunc("DELETE /api/v1/pipelines/{id}", s.deletePipeline)
 	// audit + reports routes
 	s.mux.HandleFunc("GET /api/v1/audit", s.listAudit)
 	s.mux.HandleFunc("GET /api/v1/reports", s.getReports)
@@ -358,6 +370,94 @@ func (s *Server) deleteSource(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
+// ---- pipeline handlers -----------------------------------------------------
+
+// listPipelines handles GET /api/v1/pipelines
+func (s *Server) listPipelines(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, s.pipelines.List())
+}
+
+// createPipeline handles POST /api/v1/pipelines
+func (s *Server) createPipeline(w http.ResponseWriter, r *http.Request) {
+	var p channel.Pipeline
+	if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	if p.ID == "" {
+		writeError(w, http.StatusBadRequest, "id is required")
+		return
+	}
+	if p.Name == "" {
+		writeError(w, http.StatusBadRequest, "name is required")
+		return
+	}
+	if err := s.pipelines.Add(p); err != nil {
+		if errors.Is(err, channel.ErrDuplicateID) {
+			writeError(w, http.StatusConflict, "pipeline ID already exists")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to create pipeline")
+		return
+	}
+	got, _ := s.pipelines.Get(p.ID)
+	writeJSON(w, http.StatusCreated, got)
+}
+
+// getPipeline handles GET /api/v1/pipelines/{id}
+func (s *Server) getPipeline(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	p, err := s.pipelines.Get(id)
+	if err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "pipeline not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to retrieve pipeline")
+		return
+	}
+	writeJSON(w, http.StatusOK, p)
+}
+
+// updatePipeline handles PUT /api/v1/pipelines/{id}
+func (s *Server) updatePipeline(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	var p channel.Pipeline
+	if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	p.ID = id // path wins over body
+	if p.Name == "" {
+		writeError(w, http.StatusBadRequest, "name is required")
+		return
+	}
+	if err := s.pipelines.Update(p); err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "pipeline not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to update pipeline")
+		return
+	}
+	got, _ := s.pipelines.Get(id)
+	writeJSON(w, http.StatusOK, got)
+}
+
+// deletePipeline handles DELETE /api/v1/pipelines/{id}
+func (s *Server) deletePipeline(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	if err := s.pipelines.Delete(id); err != nil {
+		if errors.Is(err, channel.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "pipeline not found")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "failed to delete pipeline")
+		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(), channel.NewSourceRegistry())
+	return rest.New(st, channel.NewRegistry(), channel.NewSourceRegistry(), channel.NewPipelineRegistry())
 }
 
 // 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, channel.NewSourceRegistry())
+	return rest.New(st, reg, channel.NewSourceRegistry(), channel.NewPipelineRegistry())
 }
 
 func get(srv http.Handler, path string) *httptest.ResponseRecorder {

internal/channel/channel.go

@@ -285,3 +285,121 @@ func (r *SourceRegistry) Len() int {
 	defer r.mu.RUnlock()
 	return len(r.sources)
 }
+
+// ---- Pipelines -------------------------------------------------------------
+
+// PipelineFilters defines message-level filter criteria applied before delivery.
+type PipelineFilters struct {
+	// EventTypes limits processing to the listed HL7 event types (e.g. "ADT^A01").
+	// An empty slice means accept all event types.
+	EventTypes []string `yaml:"event_types" json:"event_types,omitempty"`
+	// MinScore is the minimum quality score [0,1] required to proceed.
+	// Messages scoring below this threshold are discarded without delivery.
+	MinScore float64 `yaml:"min_score" json:"min_score"`
+}
+
+// Pipeline routes messages from one source through filters to one or more
+// delivery channels.
+type Pipeline struct {
+	// ID is the unique identifier for this pipeline (URL-safe string).
+	ID string `yaml:"id" json:"id"`
+	// Name is a human-readable label.
+	Name string `yaml:"name" json:"name"`
+	// SourceID is the ID of the SourceConfig that feeds this pipeline.
+	// An empty string means accept messages from any source.
+	SourceID string `yaml:"source_id" json:"source_id,omitempty"`
+	// Filters defines optional message-level criteria.
+	Filters PipelineFilters `yaml:"filters" json:"filters"`
+	// DestinationIDs is the list of Channel IDs to deliver to.
+	DestinationIDs []string `yaml:"destination_ids" json:"destination_ids,omitempty"`
+	// Enabled controls whether this pipeline participates in routing.
+	Enabled bool `yaml:"enabled" json:"enabled"`
+	// CreatedAt is set when the pipeline is first registered.
+	CreatedAt time.Time `json:"created_at"`
+	// UpdatedAt is refreshed on every update.
+	UpdatedAt time.Time `json:"updated_at"`
+}
+
+// PipelineRegistry is a thread-safe, in-memory store of Pipelines.
+type PipelineRegistry struct {
+	mu        sync.RWMutex
+	pipelines map[string]Pipeline
+	now       func() time.Time
+}
+
+// NewPipelineRegistry creates an empty PipelineRegistry.
+func NewPipelineRegistry() *PipelineRegistry {
+	return &PipelineRegistry{
+		pipelines: make(map[string]Pipeline),
+		now:       time.Now,
+	}
+}
+
+// Add inserts a new pipeline. Returns ErrDuplicateID if the ID already exists.
+func (r *PipelineRegistry) Add(p Pipeline) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if _, exists := r.pipelines[p.ID]; exists {
+		return ErrDuplicateID
+	}
+	now := r.now()
+	p.CreatedAt = now
+	p.UpdatedAt = now
+	r.pipelines[p.ID] = p
+	return nil
+}
+
+// Update replaces an existing pipeline. Returns ErrNotFound if the ID does not
+// exist. CreatedAt is preserved; UpdatedAt is refreshed.
+func (r *PipelineRegistry) Update(p Pipeline) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	existing, exists := r.pipelines[p.ID]
+	if !exists {
+		return ErrNotFound
+	}
+	p.CreatedAt = existing.CreatedAt
+	p.UpdatedAt = r.now()
+	r.pipelines[p.ID] = p
+	return nil
+}
+
+// Delete removes a pipeline by ID. Returns ErrNotFound if the ID does not exist.
+func (r *PipelineRegistry) Delete(id string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if _, exists := r.pipelines[id]; !exists {
+		return ErrNotFound
+	}
+	delete(r.pipelines, id)
+	return nil
+}
+
+// Get returns a copy of the pipeline with the given ID.
+func (r *PipelineRegistry) Get(id string) (Pipeline, error) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	p, exists := r.pipelines[id]
+	if !exists {
+		return Pipeline{}, ErrNotFound
+	}
+	return p, nil
+}
+
+// List returns a copy of all pipelines. Order is not guaranteed.
+func (r *PipelineRegistry) List() []Pipeline {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	out := make([]Pipeline, 0, len(r.pipelines))
+	for _, p := range r.pipelines {
+		out = append(out, p)
+	}
+	return out
+}
+
+// Len returns the number of registered pipelines.
+func (r *PipelineRegistry) Len() int {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return len(r.pipelines)
+}

web/src/App.tsx

@@ -3,6 +3,8 @@ import { ToastProvider } from './components/Toast'
 import Layout from './components/Layout'
 import Dashboard from './pages/Dashboard'
 import Channels from './pages/Channels'
+import Sources from './pages/Sources'
+import Pipelines from './pages/Pipelines'
 import Messages from './pages/Messages'
 import AuditLog from './pages/AuditLog'
 import Reports from './pages/Reports'
@@ -16,6 +18,8 @@ export default function App() {
           <Route index element={<Navigate to="/dashboard" replace />} />
           <Route path="dashboard" element={<Dashboard />} />
           <Route path="channels" element={<Channels />} />
+          <Route path="sources" element={<Sources />} />
+          <Route path="pipelines" element={<Pipelines />} />
           <Route path="messages" element={<Messages />} />
           <Route path="messages/:id" element={<Messages />} />
           <Route path="audit" element={<AuditLog />} />

web/src/api/client.ts

@@ -2,7 +2,7 @@
 // During local dev Vite proxies /api, /healthz, /readyz to localhost:8080.
 // In production nginx does the same proxy.
 
-import type { AuditEntry, Channel, HealthResponse, Message, MessageStatus, ReportSummary } from '../types'
+import type { AuditEntry, Channel, HealthResponse, Message, MessageStatus, Pipeline, ReportSummary, Source } from '../types'
 
 async function request<T>(path: string, init?: RequestInit): Promise<T> {
   const res = await fetch(path, {
@@ -46,6 +46,40 @@ export const api = {
       request<void>(`/api/v1/channels/${id}`, { method: 'DELETE' }),
   },
 
+  sources: {
+    list: () => request<Source[]>('/api/v1/sources'),
+    get: (id: string) => request<Source>(`/api/v1/sources/${id}`),
+    create: (src: Omit<Source, 'created_at' | 'updated_at'>) =>
+      request<Source>('/api/v1/sources', {
+        method: 'POST',
+        body: JSON.stringify(src),
+      }),
+    update: (id: string, src: Partial<Omit<Source, 'created_at' | 'updated_at'>>) =>
+      request<Source>(`/api/v1/sources/${id}`, {
+        method: 'PUT',
+        body: JSON.stringify(src),
+      }),
+    delete: (id: string) =>
+      request<void>(`/api/v1/sources/${id}`, { method: 'DELETE' }),
+  },
+
+  pipelines: {
+    list: () => request<Pipeline[]>('/api/v1/pipelines'),
+    get: (id: string) => request<Pipeline>(`/api/v1/pipelines/${id}`),
+    create: (p: Omit<Pipeline, 'created_at' | 'updated_at'>) =>
+      request<Pipeline>('/api/v1/pipelines', {
+        method: 'POST',
+        body: JSON.stringify(p),
+      }),
+    update: (id: string, p: Partial<Omit<Pipeline, 'created_at' | 'updated_at'>>) =>
+      request<Pipeline>(`/api/v1/pipelines/${id}`, {
+        method: 'PUT',
+        body: JSON.stringify(p),
+      }),
+    delete: (id: string) =>
+      request<void>(`/api/v1/pipelines/${id}`, { method: 'DELETE' }),
+  },
+
   health: {
     liveness: () => request<HealthResponse>('/healthz'),
     readiness: () => request<HealthResponse>('/readyz'),