Add delivery, retry attempts, DLQ, and persistence

Enable real delivery and observability: introduce DLQ wiring, per-channel Sender abstraction, and persistence of routed payloads.

Highlights:
- cmd/gateway: wire a DLQ writer and Store into the router/pipeline; persist messages before delivery and update stored status after routing.
- internal/config: add DLQDir config with default.
- internal/audit: include Attempts in audit entries.
- internal/connector/destination/retry: add SendWithAttempts to report actual attempt counts (Send delegates to it).
- internal/router: implement Sender/SenderBuilder, retry-backed default builder, per-channel sender cache with invalidation, delivery handling with attempt/duration reporting, DLQ writes on terminal failures, and AggregateStatus to summarize per-message outcome.
- Tests: expand router tests to cover real delivery, DLQ writing, attempt counts, sender cache behavior, and status aggregation.
- web UI/types: surface new stages ('deliver','store') and status/attempts badges; add attempts to AuditEntry type.

Purpose: this change wires up end-to-end delivery, records attempts for auditing/metrics, persists message state for the UI, and dead-letters permanently failing deliveries for operator inspection.

Author: vagnercazarotto

cmd/gateway/main.go

@@ -14,12 +14,14 @@ import (
 	"github.com/vagnercazarotto/verifhir-gateway/internal/audit"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/channel"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/config"
+	"github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/dlq"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/ingest/mllp"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/mapping"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/parser"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/quality"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/router"
+	"github.com/vagnercazarotto/verifhir-gateway/internal/store"
 	"github.com/vagnercazarotto/verifhir-gateway/internal/store/sqlite"
 )
 
@@ -49,9 +51,12 @@ func main() {
 		}
 	}
 
-	// Channel-aware dispatcher (Phase 3 PR 1: scaffold only — real HTTP
-	// delivery, retry and DLQ wiring follow in subsequent PRs).
-	rtr := router.New(reg)
+	// Dead-letter writer for delivery failures (one shared dir for the gateway).
+	dlqW := dlq.New(dlq.Config{Dir: cfg.DLQDir})
+
+	// Channel-aware dispatcher with real HTTP delivery, per-channel retry,
+	// and DLQ on terminal failure.
+	rtr := router.New(reg, dlqW)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
@@ -69,7 +74,7 @@ func main() {
 	}()
 
 	// MLLP listener (blocks until ctx cancelled)
-	mllpSrv := mllp.New(cfg.MLLPAddr, makePipeline(rtr))
+	mllpSrv := mllp.New(cfg.MLLPAddr, makePipeline(rtr, st))
 	if err := mllpSrv.ListenAndServe(ctx); err != nil {
 		log.Printf("[gateway] mllp: %v", err)
 	}
@@ -87,8 +92,10 @@ func main() {
 // processing chain and emits one structured audit log line per stage.
 //
 // Routing is delegated to rtr, which fans the payload out across the
-// channels currently registered (see internal/router).
-func makePipeline(rtr *router.Router) func(model.HL7Message) error {
+// channels currently registered (see internal/router). Persistence is
+// delegated to st: every message is saved before routing and the aggregated
+// delivery outcome is recorded after the router returns.
+func makePipeline(rtr *router.Router, st store.Store) func(model.HL7Message) error {
 	return func(msg model.HL7Message) error {
 		audit.Log(audit.Entry{
 			MessageID: msg.ID,
@@ -142,15 +149,45 @@ func makePipeline(rtr *router.Router) func(model.HL7Message) error {
 			Findings:     len(report.Findings),
 		})
 
+		payload := model.RoutedPayload{Resource: resource, Quality: report}
+
+		// Persist as pending before delivery so the message is visible in
+		// the UI even if routing crashes or the gateway restarts mid-flight.
+		ctx := context.Background()
+		t = time.Now()
+		if err := st.Save(ctx, payload); err != nil {
+			audit.Log(audit.Entry{
+				MessageID:  msg.ID,
+				Stage:      "store",
+				DurationMs: time.Since(t).Milliseconds(),
+				Status:     "error",
+				Error:      err.Error(),
+			})
+			// Continue: persistence failure must not block delivery.
+		}
+
 		t = time.Now()
-		rtr.Route(context.Background(), model.RoutedPayload{Resource: resource, Quality: report})
+		decisions := rtr.Route(ctx, payload)
 		audit.Log(audit.Entry{
 			MessageID:  msg.ID,
 			Stage:      "route",
 			DurationMs: time.Since(t).Milliseconds(),
 			Status:     "ok",
 		})
 
+		// Reflect the aggregated delivery outcome on the stored record.
+		status, attempts, lastErr := router.AggregateStatus(decisions)
+		t = time.Now()
+		if err := st.UpdateStatus(ctx, msg.ID, status, attempts, lastErr); err != nil {
+			audit.Log(audit.Entry{
+				MessageID:  msg.ID,
+				Stage:      "store",
+				DurationMs: time.Since(t).Milliseconds(),
+				Status:     "error",
+				Error:      err.Error(),
+			})
+		}
+
 		return nil
 	}
 }

internal/audit/logger.go

@@ -35,6 +35,8 @@ type Entry struct {
 	ChannelID string `json:"channel_id,omitempty"`
 	// DestURL is the resolved target URL for the delivery attempt.
 	DestURL string `json:"dest_url,omitempty"`
+	// Attempts is the number of delivery tries actually executed.
+	Attempts int `json:"attempts,omitempty"`
 }
 
 // Log writes e as a single JSON line to the audit sink.

internal/config/config.go

@@ -11,6 +11,7 @@ type Config struct {
 	DBPath       string // SQLite file path; ignored when DATABASE_URL is set
 	ChannelsFile string // optional YAML file with channel definitions
 	AuditDir     string // directory for daily .jsonl audit files
+	DLQDir       string // directory for dead-letter JSON files
 }
 
 func Load() Config {
@@ -34,11 +35,17 @@ func Load() Config {
 		auditDir = ".local/audit"
 	}
 
+	dlqDir := os.Getenv("GATEWAY_DLQ_DIR")
+	if dlqDir == "" {
+		dlqDir = ".local/dlq"
+	}
+
 	return Config{
 		HTTPPort:     port,
 		MLLPAddr:     mllp,
 		DBPath:       dbPath,
 		ChannelsFile: os.Getenv("GATEWAY_CHANNELS_FILE"),
 		AuditDir:     auditDir,
+		DLQDir:       dlqDir,
 	}
 }

internal/connector/destination/retry/retryer.go

@@ -67,13 +67,25 @@ func (r *Retryer) SetSleep(fn func(ctx context.Context, d time.Duration) error)
 // It returns the last error if all attempts are exhausted, or the context
 // error if the context is cancelled between retries.
 func (r *Retryer) Send(ctx context.Context, payload model.RoutedPayload) error {
+	_, err := r.SendWithAttempts(ctx, payload)
+	return err
+}
+
+// SendWithAttempts is like Send but also returns the number of attempts that
+// were actually executed (1-based). Callers that need to record the attempt
+// count for auditing or metrics should prefer this method.
+//
+// On success the returned count reflects the attempt that succeeded
+// (e.g. 1 if the first try worked). On failure or context cancellation it
+// reflects the last attempt that ran.
+func (r *Retryer) SendWithAttempts(ctx context.Context, payload model.RoutedPayload) (int, error) {
 	backoff := r.cfg.InitialBackoff
 	var lastErr error
 
 	for attempt := 0; attempt < r.cfg.MaxAttempts; attempt++ {
 		lastErr = r.inner.Send(ctx, payload)
 		if lastErr == nil {
-			return nil
+			return attempt + 1, nil
 		}
 
 		// Last attempt — do not sleep.
@@ -82,12 +94,12 @@ func (r *Retryer) Send(ctx context.Context, payload model.RoutedPayload) error {
 		}
 
 		if err := r.sleep(ctx, backoff); err != nil {
-			return fmt.Errorf("retry: context cancelled after %d attempt(s): %w", attempt+1, err)
+			return attempt + 1, fmt.Errorf("retry: context cancelled after %d attempt(s): %w", attempt+1, err)
 		}
 		backoff = time.Duration(float64(backoff) * r.cfg.Multiplier)
 	}
 
-	return fmt.Errorf("retry: all %d attempt(s) failed: %w", r.cfg.MaxAttempts, lastErr)
+	return r.cfg.MaxAttempts, fmt.Errorf("retry: all %d attempt(s) failed: %w", r.cfg.MaxAttempts, lastErr)
 }
 
 func sleepWithContext(ctx context.Context, d time.Duration) error {