Add MLLP output option and adapter

Add support for HL7 passthrough over MLLP: new mllp adapter sends raw HL7v2 framed with VT/FS/CR, reads ACKs and treats MSA AE/AR as NACKs (includes unit tests). Persist RawHL7 on model.RoutedPayload and include it in the gateway pipeline. Introduce Channel.OutputType ("fhir" | "hl7_passthrough") and update the router to construct either an HTTP or MLLP sender (both wrapped with retry). Update channel model, API types, and the web UI to let users pick output_type, adjust placeholders and form fields (hide auth for MLLP), and show output type in the channels list.

Author: vagnercazarotto

cmd/gateway/main.go

@@ -149,7 +149,7 @@ func makePipeline(rtr *router.Router, st store.Store) func(model.HL7Message) err
 			Findings:     len(report.Findings),
 		})
 
-		payload := model.RoutedPayload{Resource: resource, Quality: report}
+		payload := model.RoutedPayload{Resource: resource, Quality: report, RawHL7: msg.Payload}
 
 		// Persist as pending before delivery so the message is visible in
 		// the UI even if routing crashes or the gateway restarts mid-flight.

internal/channel/channel.go

@@ -33,17 +33,31 @@ type RetryConfig struct {
 	Multiplier float64 `yaml:"multiplier" json:"multiplier"`
 }
 
-// Channel is a named FHIR delivery destination with its own delivery policy.
+// OutputType controls what format/protocol is used when delivering messages.
+type OutputType string
+
+const (
+	// OutputFHIR delivers the mapped FHIR Bundle as JSON via HTTP POST (default).
+	OutputFHIR OutputType = "fhir"
+	// OutputHL7Passthrough delivers the original raw HL7v2 message via MLLP TCP.
+	OutputHL7Passthrough OutputType = "hl7_passthrough"
+)
+
+// Channel is a named delivery destination with its own delivery policy.
 type Channel struct {
 	// ID is the unique identifier for this channel (URL-safe string).
 	ID string `yaml:"id" json:"id"`
 	// Name is a human-readable label.
 	Name string `yaml:"name" json:"name"`
-	// URL is the target FHIR server endpoint.
+	// OutputType controls the output format and protocol. Defaults to "fhir".
+	OutputType OutputType `yaml:"output_type" json:"output_type"`
+	// URL is the target endpoint.
+	// For fhir: HTTP/HTTPS URL of the FHIR server.
+	// For hl7_passthrough: host:port of the MLLP destination (e.g. "hl7.example.com:2575").
 	URL string `yaml:"url" json:"url"`
-	// AuthHeader is the value sent in the Authorization header (optional).
+	// AuthHeader is the value sent in the Authorization header (optional, FHIR only).
 	AuthHeader string `yaml:"auth_header" json:"auth_header,omitempty"`
-	// TimeoutMS is the HTTP client timeout in milliseconds (default 10 000).
+	// TimeoutMS is the client timeout in milliseconds (default 10 000).
 	TimeoutMS int `yaml:"timeout_ms" json:"timeout_ms"`
 	// MinQualityScore is the minimum quality score [0,1] required to deliver.
 	// Messages scoring below this threshold are dead-lettered without sending.

internal/connector/destination/mllp/adapter.go

@@ -0,0 +1,156 @@
+// Package mllp provides an MLLP destination adapter that delivers the original
+// raw HL7v2 message to a remote system over a TCP connection using MLLP framing.
+//
+// The adapter opens a new TCP connection per Send call so it is safe for
+// concurrent use and requires no persistent connection management. For
+// high-throughput scenarios a connection pool can be added later without
+// changing the Sender interface.
+//
+// MLLP framing:
+//
+//	VT (0x0B) <HL7 payload bytes> FS (0x1C) CR (0x0D)
+package mllp
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"time"
+
+	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
+)
+
+const (
+	vt              = byte(0x0B)
+	fs              = byte(0x1C)
+	cr              = byte(0x0D)
+	defaultTimeout  = 10 * time.Second
+	ackBufSize      = 4096
+)
+
+// Config holds the runtime settings for the MLLP destination adapter.
+type Config struct {
+	// Addr is the host:port of the remote MLLP listener, e.g. "hl7.example.com:2575".
+	Addr string
+
+	// Timeout overrides the default 10-second dial+write+ack deadline.
+	// Zero means use the default.
+	Timeout time.Duration
+}
+
+// Adapter sends raw HL7v2 messages to a remote MLLP listener.
+type Adapter struct {
+	cfg     Config
+	timeout time.Duration
+	// dial is injectable for testing; defaults to net.DialTimeout.
+	dial func(network, addr string, timeout time.Duration) (net.Conn, error)
+}
+
+// New creates an Adapter from cfg.
+func New(cfg Config) *Adapter {
+	t := cfg.Timeout
+	if t <= 0 {
+		t = defaultTimeout
+	}
+	return &Adapter{
+		cfg:     cfg,
+		timeout: t,
+		dial:    net.DialTimeout,
+	}
+}
+
+// Send wraps payload.RawHL7 in MLLP framing and delivers it to the configured
+// address, then reads and validates the ACK response.
+// It returns an error if RawHL7 is empty, the connection fails, or the remote
+// side responds with a NACK (MSA-1 = "AE" or "AR").
+func (a *Adapter) Send(ctx context.Context, payload model.RoutedPayload) error {
+	raw := payload.RawHL7
+	if raw == "" {
+		return fmt.Errorf("mllp adapter: RawHL7 is empty for msg %s", payload.Resource.ID)
+	}
+
+	// Respect context deadline.
+	deadline := time.Now().Add(a.timeout)
+	if d, ok := ctx.Deadline(); ok && d.Before(deadline) {
+		deadline = d
+	}
+
+	conn, err := a.dial("tcp", a.cfg.Addr, time.Until(deadline))
+	if err != nil {
+		return fmt.Errorf("mllp adapter: dial %s: %w", a.cfg.Addr, err)
+	}
+	defer conn.Close()
+
+	if err := conn.SetDeadline(deadline); err != nil {
+		return fmt.Errorf("mllp adapter: set deadline: %w", err)
+	}
+
+	// Write MLLP-framed message.
+	frame := make([]byte, 0, 3+len(raw)+2)
+	frame = append(frame, vt)
+	frame = append(frame, []byte(raw)...)
+	frame = append(frame, fs, cr)
+
+	if _, err := conn.Write(frame); err != nil {
+		return fmt.Errorf("mllp adapter: write: %w", err)
+	}
+
+	// Read ACK response.
+	ack, err := readACK(conn)
+	if err != nil {
+		return fmt.Errorf("mllp adapter: read ack: %w", err)
+	}
+	if isNACK(ack) {
+		return fmt.Errorf("mllp adapter: received NACK from %s: %s", a.cfg.Addr, ack)
+	}
+	return nil
+}
+
+// readACK reads an MLLP-framed response from conn and returns the payload.
+func readACK(conn net.Conn) (string, error) {
+	buf := make([]byte, 1)
+	// Advance to VT.
+	for {
+		if _, err := io.ReadFull(conn, buf); err != nil {
+			return "", err
+		}
+		if buf[0] == vt {
+			break
+		}
+	}
+	var payload []byte
+	for len(payload) < ackBufSize {
+		if _, err := io.ReadFull(conn, buf); err != nil {
+			return "", err
+		}
+		if buf[0] == fs {
+			break
+		}
+		payload = append(payload, buf[0])
+	}
+	// Consume trailing CR (best-effort; ignore error).
+	_, _ = io.ReadFull(conn, buf)
+	return string(payload), nil
+}
+
+// isNACK returns true when the ACK message contains MSA-1 = "AE" or "AR".
+// This is a minimal check on the MSA segment field 1.
+func isNACK(ack string) bool {
+	// Look for MSA segment and read field 1 (after first |).
+	const prefix = "MSA|"
+	idx := 0
+	for i := 0; i+len(prefix) <= len(ack); i++ {
+		if ack[i:i+len(prefix)] == prefix {
+			idx = i + len(prefix)
+			// Field 1 is the next 2 characters before the next |.
+			end := idx
+			for end < len(ack) && ack[end] != '|' && ack[end] != '\r' && ack[end] != '\n' {
+				end++
+			}
+			code := ack[idx:end]
+			return code == "AE" || code == "AR"
+		}
+	}
+	return false
+}

internal/connector/destination/mllp/adapter_test.go

@@ -0,0 +1,137 @@
+package mllp
+
+import (
+	"context"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
+)
+
+// ---- helpers ---------------------------------------------------------------
+
+func makeACK(code string) []byte {
+	ack := "MSH|^~\\&|TEST|TEST|SENDER|SENDER|20260101120000||ACK^A01|1|P|2.5\rMSA|" + code + "|1\r"
+	out := make([]byte, 0, len(ack)+3)
+	out = append(out, vt)
+	out = append(out, []byte(ack)...)
+	out = append(out, fs, cr)
+	return out
+}
+
+func serveOnce(t *testing.T, response []byte) (addr string, done <-chan struct{}) {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	ch := make(chan struct{})
+	go func() {
+		defer close(ch)
+		defer ln.Close()
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		buf := make([]byte, 4096)
+		conn.Read(buf) // consume inbound frame
+		conn.Write(response)
+	}()
+	return ln.Addr().String(), ch
+}
+
+func samplePayload(raw string) model.RoutedPayload {
+	return model.RoutedPayload{
+		Resource: model.FHIRResource{ID: "msg-001"},
+		RawHL7:   raw,
+	}
+}
+
+const sampleHL7 = "MSH|^~\\&|SEND|SEND|RECV|RECV|20260101120000||ADT^A01|1|P|2.5\r" +
+	"PID|1||P001^^^MRN||Doe^John\r"
+
+// ---- tests -----------------------------------------------------------------
+
+func TestSendACK(t *testing.T) {
+	addr, done := serveOnce(t, makeACK("AA"))
+	a := New(Config{Addr: addr, Timeout: 3 * time.Second})
+	if err := a.Send(context.Background(), samplePayload(sampleHL7)); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	<-done
+}
+
+func TestSendNACK_AE(t *testing.T) {
+	addr, done := serveOnce(t, makeACK("AE"))
+	a := New(Config{Addr: addr, Timeout: 3 * time.Second})
+	err := a.Send(context.Background(), samplePayload(sampleHL7))
+	if err == nil {
+		t.Fatal("expected error for AE NACK")
+	}
+	if !strings.Contains(err.Error(), "NACK") {
+		t.Errorf("error should mention NACK, got: %v", err)
+	}
+	<-done
+}
+
+func TestSendNACK_AR(t *testing.T) {
+	addr, done := serveOnce(t, makeACK("AR"))
+	a := New(Config{Addr: addr, Timeout: 3 * time.Second})
+	err := a.Send(context.Background(), samplePayload(sampleHL7))
+	if err == nil {
+		t.Fatal("expected error for AR NACK")
+	}
+	<-done
+}
+
+func TestSendEmptyRawHL7(t *testing.T) {
+	a := New(Config{Addr: "127.0.0.1:9999", Timeout: time.Second})
+	err := a.Send(context.Background(), samplePayload(""))
+	if err == nil {
+		t.Fatal("expected error when RawHL7 is empty")
+	}
+	if !strings.Contains(err.Error(), "RawHL7 is empty") {
+		t.Errorf("unexpected error: %v", err)
+	}
+}
+
+func TestSendDialFailure(t *testing.T) {
+	// Port 1 is reserved and will always fail to connect.
+	a := New(Config{Addr: "127.0.0.1:1", Timeout: 500 * time.Millisecond})
+	err := a.Send(context.Background(), samplePayload(sampleHL7))
+	if err == nil {
+		t.Fatal("expected dial error")
+	}
+}
+
+func TestSendContextCancelled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // already cancelled
+	// Point at a non-routable address so the dial races the deadline.
+	a := New(Config{Addr: "192.0.2.1:2575", Timeout: 100 * time.Millisecond})
+	err := a.Send(ctx, samplePayload(sampleHL7))
+	if err == nil {
+		t.Fatal("expected error with cancelled context")
+	}
+}
+
+func TestIsNACK(t *testing.T) {
+	cases := []struct {
+		ack  string
+		want bool
+	}{
+		{"MSH|...\rMSA|AA|1\r", false},
+		{"MSH|...\rMSA|AE|1\r", true},
+		{"MSH|...\rMSA|AR|1\r", true},
+		{"MSH|...\rMSA|CA|1\r", false},
+		{"", false},
+	}
+	for _, c := range cases {
+		if got := isNACK(c.ack); got != c.want {
+			t.Errorf("isNACK(%q) = %v, want %v", c.ack, got, c.want)
+		}
+	}
+}

internal/model/types.go

@@ -199,4 +199,7 @@ type MDMResult struct {
 type RoutedPayload struct {
 	Resource FHIRResource
 	Quality  QualityReport
+	// RawHL7 holds the original HL7v2 message text. It is populated by the
+	// ingest stage and used by hl7_passthrough channels.
+	RawHL7 string
 }

internal/router/router.go

@@ -15,6 +15,7 @@ import (
 	"github.com/vagnercazarotto/verifhir-gateway/internal/channel"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/dlq"
 	httpdest "github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/http"
+	mllpdest "github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/mllp"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/retry"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
 )
@@ -178,14 +179,25 @@ func (r *Router) senderFor(ch channel.Channel) Sender {
 	return sender
 }
 
-// defaultBuilder constructs a retry-wrapped HTTP adapter from a Channel.
+// defaultBuilder constructs a retry-wrapped sender from a Channel.
+// For fhir channels it wraps an HTTP adapter; for hl7_passthrough channels
+// it wraps an MLLP client adapter.
 func defaultBuilder(ch channel.Channel) Sender {
-	adapter := httpdest.New(httpdest.Config{
-		URL:        ch.URL,
-		Timeout:    ch.Timeout(),
-		AuthHeader: ch.AuthHeader,
-	})
-	return retry.New(adapter, retry.Config{
+	var inner retry.Sender
+	switch ch.OutputType {
+	case channel.OutputHL7Passthrough:
+		inner = mllpdest.New(mllpdest.Config{
+			Addr:    ch.URL,
+			Timeout: ch.Timeout(),
+		})
+	default: // OutputFHIR and any unrecognised value
+		inner = httpdest.New(httpdest.Config{
+			URL:        ch.URL,
+			Timeout:    ch.Timeout(),
+			AuthHeader: ch.AuthHeader,
+		})
+	}
+	return retry.New(inner, retry.Config{
 		MaxAttempts:    ch.Retry.MaxAttempts,
 		InitialBackoff: time.Duration(ch.Retry.InitialBackoffMS) * time.Millisecond,
 		Multiplier:     ch.Retry.Multiplier,