Add DLQ writer and retryer with tests

Introduce a dead-letter queue writer (internal/connector/destination/dlq) that persists failed RoutedPayloads as timestamped JSON files. Files include timestamp, message ID, error, attempts and the original payload; the directory is auto-created and filenames are sanitized. Add a retryer (internal/connector/destination/retry) that wraps a Sender with configurable MaxAttempts, InitialBackoff and Multiplier to perform exponential-backoff retries, respects context cancellation, and exposes injectable sleep/now hooks for tests. Unit tests added for both components to validate directory/file creation, JSON contents, filename format, retry behaviour, backoff growth and defaulting.

Author: vagnercazarotto

internal/connector/destination/dlq/writer.go

@@ -0,0 +1,94 @@
+// Package dlq implements a dead-letter queue that persists failed delivery
+// payloads to the local filesystem as JSON files.
+//
+// Each failed message is written to Dir as a timestamped JSON file containing
+// the full RoutedPayload, the error message, and the number of delivery
+// attempts. Files can be replayed manually or by an automated recovery process.
+package dlq
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
+)
+
+// Entry is the JSON structure written for each dead-lettered message.
+type Entry struct {
+	Timestamp string              `json:"ts"`
+	MessageID string              `json:"msg_id"`
+	Error     string              `json:"error"`
+	Attempts  int                 `json:"attempts"`
+	Payload   model.RoutedPayload `json:"payload"`
+}
+
+// Config controls where dead-letter files are written.
+type Config struct {
+	// Dir is the directory for dead-letter JSON files.
+	// It is created automatically if it does not exist.
+	Dir string
+}
+
+// Writer persists failed payloads to disk.
+type Writer struct {
+	cfg Config
+	now func() time.Time // injectable for tests
+}
+
+// New creates a Writer. The dead-letter directory is not created until the
+// first Write call.
+func New(cfg Config) *Writer {
+	return &Writer{cfg: cfg, now: time.Now}
+}
+
+// SetNow replaces the time source. Used only in tests.
+func (w *Writer) SetNow(fn func() time.Time) { w.now = fn }
+
+// Write serialises payload as a JSON dead-letter entry and writes it to Dir.
+// The file name is <unix-nano>-<msgID>.json, guaranteeing uniqueness.
+// It returns an error only if the file cannot be written; callers should log
+// this but must not block the pipeline on a write failure.
+func (w *Writer) Write(payload model.RoutedPayload, attempts int, cause error) error {
+	if err := os.MkdirAll(w.cfg.Dir, 0o750); err != nil {
+		return fmt.Errorf("dlq: create dir: %w", err)
+	}
+
+	ts := w.now()
+	entry := Entry{
+		Timestamp: ts.UTC().Format(time.RFC3339),
+		MessageID: payload.Resource.ID,
+		Error:     cause.Error(),
+		Attempts:  attempts,
+		Payload:   payload,
+	}
+
+	data, err := json.MarshalIndent(entry, "", "  ")
+	if err != nil {
+		return fmt.Errorf("dlq: marshal: %w", err)
+	}
+
+	name := fmt.Sprintf("%d-%s.json", ts.UnixNano(), sanitize(payload.Resource.ID))
+	path := filepath.Join(w.cfg.Dir, name)
+	if err := os.WriteFile(path, data, 0o640); err != nil {
+		return fmt.Errorf("dlq: write file: %w", err)
+	}
+	return nil
+}
+
+// sanitize removes characters unsafe for filenames.
+func sanitize(s string) string {
+	out := make([]byte, 0, len(s))
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
+			(c >= '0' && c <= '9') || c == '-' || c == '_' {
+			out = append(out, c)
+		} else {
+			out = append(out, '_')
+		}
+	}
+	return string(out)
+}

internal/connector/destination/dlq/writer_test.go

@@ -0,0 +1,169 @@
+package dlq_test
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/vagnercazarotto/verifhir-gateway/internal/connector/destination/dlq"
+	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
+)
+
+var testPayload = model.RoutedPayload{
+	Resource: model.FHIRResource{ID: "msg-abc"},
+}
+
+func newWriter(dir string, t time.Time) *dlq.Writer {
+	w := dlq.New(dlq.Config{Dir: dir})
+	w.SetNow(func() time.Time { return t })
+	return w
+}
+
+func TestCreatesDirectory(t *testing.T) {
+	dir := filepath.Join(t.TempDir(), "sub", "dlq")
+	w := newWriter(dir, time.Now())
+
+	if err := w.Write(testPayload, 1, errors.New("fail")); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		t.Error("directory was not created")
+	}
+}
+
+func TestWritesValidJSONFile(t *testing.T) {
+	dir := t.TempDir()
+	w := newWriter(dir, time.Now())
+
+	if err := w.Write(testPayload, 2, errors.New("connection refused")); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("read dir: %v", err)
+	}
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 file, got %d", len(entries))
+	}
+
+	data, err := os.ReadFile(filepath.Join(dir, entries[0].Name()))
+	if err != nil {
+		t.Fatalf("read file: %v", err)
+	}
+
+	var entry dlq.Entry
+	if err := json.Unmarshal(data, &entry); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+}
+
+func TestFileContainsMessageID(t *testing.T) {
+	dir := t.TempDir()
+	w := newWriter(dir, time.Now())
+	_ = w.Write(testPayload, 1, errors.New("fail"))
+
+	entries, _ := os.ReadDir(dir)
+	data, _ := os.ReadFile(filepath.Join(dir, entries[0].Name()))
+
+	var entry dlq.Entry
+	_ = json.Unmarshal(data, &entry)
+
+	if entry.MessageID != testPayload.Resource.ID {
+		t.Errorf("msg_id: want %s, got %s", testPayload.Resource.ID, entry.MessageID)
+	}
+}
+
+func TestFileContainsError(t *testing.T) {
+	dir := t.TempDir()
+	w := newWriter(dir, time.Now())
+	_ = w.Write(testPayload, 1, errors.New("something went wrong"))
+
+	entries, _ := os.ReadDir(dir)
+	data, _ := os.ReadFile(filepath.Join(dir, entries[0].Name()))
+
+	var entry dlq.Entry
+	_ = json.Unmarshal(data, &entry)
+
+	if entry.Error != "something went wrong" {
+		t.Errorf("error field: want %q, got %q", "something went wrong", entry.Error)
+	}
+}
+
+func TestFileContainsAttemptCount(t *testing.T) {
+	dir := t.TempDir()
+	w := newWriter(dir, time.Now())
+	_ = w.Write(testPayload, 7, errors.New("fail"))
+
+	entries, _ := os.ReadDir(dir)
+	data, _ := os.ReadFile(filepath.Join(dir, entries[0].Name()))
+
+	var entry dlq.Entry
+	_ = json.Unmarshal(data, &entry)
+
+	if entry.Attempts != 7 {
+		t.Errorf("attempts: want 7, got %d", entry.Attempts)
+	}
+}
+
+func TestMultipleWritesCreateMultipleFiles(t *testing.T) {
+	dir := t.TempDir()
+	base := time.Now()
+
+	for i := 0; i < 5; i++ {
+		p := model.RoutedPayload{
+			Resource: model.FHIRResource{ID: fmt.Sprintf("msg-%d", i)},
+		}
+		w := newWriter(dir, base.Add(time.Duration(i)*time.Second))
+		if err := w.Write(p, 1, errors.New("fail")); err != nil {
+			t.Fatalf("write %d: %v", i, err)
+		}
+	}
+
+	entries, _ := os.ReadDir(dir)
+	if len(entries) != 5 {
+		t.Errorf("expected 5 files, got %d", len(entries))
+	}
+}
+
+func TestFileNameContainsMsgID(t *testing.T) {
+	dir := t.TempDir()
+	ts := time.Date(2024, 1, 15, 12, 0, 0, 0, time.UTC)
+	w := newWriter(dir, ts)
+	p := model.RoutedPayload{Resource: model.FHIRResource{ID: "my-msg-123"}}
+	_ = w.Write(p, 1, errors.New("fail"))
+
+	entries, _ := os.ReadDir(dir)
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 file, got %d", len(entries))
+	}
+	name := entries[0].Name()
+	if !strings.Contains(name, "my-msg-123") {
+		t.Errorf("file name %q should contain msg ID", name)
+	}
+	if !strings.HasSuffix(name, ".json") {
+		t.Errorf("file name %q should end with .json", name)
+	}
+}
+
+func TestTimestampInRFC3339(t *testing.T) {
+	dir := t.TempDir()
+	ts := time.Date(2024, 6, 20, 10, 30, 0, 0, time.UTC)
+	w := newWriter(dir, ts)
+	_ = w.Write(testPayload, 1, errors.New("fail"))
+
+	entries, _ := os.ReadDir(dir)
+	data, _ := os.ReadFile(filepath.Join(dir, entries[0].Name()))
+
+	var entry dlq.Entry
+	_ = json.Unmarshal(data, &entry)
+
+	if entry.Timestamp != "2024-06-20T10:30:00Z" {
+		t.Errorf("timestamp: want 2024-06-20T10:30:00Z, got %s", entry.Timestamp)
+	}
+}

internal/connector/destination/retry/retryer.go

@@ -0,0 +1,100 @@
+// Package retry wraps any Sender with exponential-backoff retry logic.
+//
+// The retryer calls the inner Sender up to MaxAttempts times. Between failures
+// it waits InitialBackoff * Multiplier^attempt before trying again. If the
+// context is cancelled during a wait the attempt loop stops immediately and
+// the context error is returned.
+package retry
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/vagnercazarotto/verifhir-gateway/internal/model"
+)
+
+// Sender is the interface that the retryer wraps.
+type Sender interface {
+	Send(ctx context.Context, payload model.RoutedPayload) error
+}
+
+// Config controls retry behaviour.
+type Config struct {
+	// MaxAttempts is the total number of send attempts (1 = no retry).
+	MaxAttempts int
+
+	// InitialBackoff is the wait duration after the first failure.
+	InitialBackoff time.Duration
+
+	// Multiplier scales the backoff after each subsequent failure.
+	// 2.0 means: 1s → 2s → 4s → 8s …
+	Multiplier float64
+}
+
+// Retryer wraps a Sender and retries on failure with exponential backoff.
+type Retryer struct {
+	inner Sender
+	cfg   Config
+	// sleep is injectable for testing; defaults to sleepWithContext.
+	sleep func(ctx context.Context, d time.Duration) error
+}
+
+// New creates a Retryer with the given inner Sender and Config.
+// If MaxAttempts < 1 it is set to 1 (try once, no retry).
+// If Multiplier <= 0 it is set to 2.0.
+func New(inner Sender, cfg Config) *Retryer {
+	if cfg.MaxAttempts < 1 {
+		cfg.MaxAttempts = 1
+	}
+	if cfg.Multiplier <= 0 {
+		cfg.Multiplier = 2.0
+	}
+	return &Retryer{
+		inner: inner,
+		cfg:   cfg,
+		sleep: sleepWithContext,
+	}
+}
+
+// SetSleep replaces the sleep implementation. It is used only in tests to
+// avoid real timer waits.
+func (r *Retryer) SetSleep(fn func(ctx context.Context, d time.Duration) error) {
+	r.sleep = fn
+}
+
+// Send calls the inner Sender, retrying up to MaxAttempts times on failure.
+// 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 {
+	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
+		}
+
+		// Last attempt — do not sleep.
+		if attempt == r.cfg.MaxAttempts-1 {
+			break
+		}
+
+		if err := r.sleep(ctx, backoff); err != nil {
+			return 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)
+}
+
+func sleepWithContext(ctx context.Context, d time.Duration) error {
+	select {
+	case <-time.After(d):
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}