feat(batch): add batch payment processing

internal/batch/batch.go

@@ -0,0 +1,106 @@
+// Package batch processes multiple payment items submitted together as a
+// single logical request, mirroring the payout batch chunking described in
+// the architecture doc without depending on the Stellar transaction layer.
+package batch
+
+import (
+	"errors"
+	"fmt"
+)
+
+// MaxItems is the largest number of payment items PayStream will accept in a
+// single batch, matching the Stellar multi-operation transaction limit.
+const MaxItems = 100
+
+// ItemStatus is the outcome of processing a single item within a batch.
+type ItemStatus string
+
+const (
+	ItemStatusAccepted ItemStatus = "accepted"
+	ItemStatusRejected ItemStatus = "rejected"
+)
+
+var (
+	// ErrEmptyBatch is returned when a batch has no items.
+	ErrEmptyBatch = errors.New("batch: at least one item is required")
+	// ErrTooManyItems is returned when a batch exceeds MaxItems.
+	ErrTooManyItems = errors.New("batch: exceeds maximum item count")
+	// ErrDuplicateIdempotencyKey is returned when a batch is submitted twice
+	// with the same idempotency key.
+	ErrDuplicateIdempotencyKey = errors.New("batch: idempotency key already used")
+)
+
+// Item is a single payment within a batch request.
+type Item struct {
+	RecipientID string
+	Amount      float64
+}
+
+// ItemResult is the outcome of validating a single Item.
+type ItemResult struct {
+	Item   Item
+	Status ItemStatus
+	Reason string
+}
+
+// Result is the outcome of processing a Batch.
+type Result struct {
+	ID      string
+	Items   []ItemResult
+	Total   float64
+	Rejects int
+}
+
+// Service validates and processes payment batches, deduping repeated
+// submissions by idempotency key.
+type Service struct {
+	seen map[string]Result
+}
+
+// NewService returns an empty batch Service.
+func NewService() *Service {
+	return &Service{seen: make(map[string]Result)}
+}
+
+// Process validates every item in a batch and returns a per-item Result. A
+// batch that has already been processed under the same idempotencyKey
+// returns the original Result rather than reprocessing.
+func (s *Service) Process(idempotencyKey, batchID string, items []Item) (Result, error) {
+	if existing, ok := s.seen[idempotencyKey]; ok {
+		if existing.ID != batchID {
+			return Result{}, fmt.Errorf("%w: %q", ErrDuplicateIdempotencyKey, idempotencyKey)
+		}
+		return existing, nil
+	}
+
+	if len(items) == 0 {
+		return Result{}, ErrEmptyBatch
+	}
+	if len(items) > MaxItems {
+		return Result{}, fmt.Errorf("%w: %d items, max %d", ErrTooManyItems, len(items), MaxItems)
+	}
+
+	res := Result{ID: batchID, Items: make([]ItemResult, 0, len(items))}
+	for _, it := range items {
+		ir := ItemResult{Item: it, Status: ItemStatusAccepted}
+		switch {
+		case it.RecipientID == "":
+			ir.Status = ItemStatusRejected
+			ir.Reason = "recipient ID is required"
+		case it.Amount <= 0:
+			ir.Status = ItemStatusRejected
+			ir.Reason = "amount must be greater than 0"
+		default:
+			res.Total += it.Amount
+		}
+		if ir.Status == ItemStatusRejected {
+			res.Rejects++
+		}
+		res.Items = append(res.Items, ir)
+	}
+
+	if idempotencyKey != "" {
+		s.seen[idempotencyKey] = res
+	}
+	return res, nil
+}

internal/batch/batch_test.go

@@ -0,0 +1,103 @@
+package batch
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestProcess_acceptsValidItems(t *testing.T) {
+	s := NewService()
+	items := []Item{
+		{RecipientID: "rec_1", Amount: 100},
+		{RecipientID: "rec_2", Amount: 250.50},
+	}
+
+	res, err := s.Process("idem_1", "bat_1", items)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.Rejects != 0 {
+		t.Fatalf("expected 0 rejects, got %d", res.Rejects)
+	}
+	if res.Total != 350.50 {
+		t.Fatalf("expected total 350.50, got %v", res.Total)
+	}
+	for _, ir := range res.Items {
+		if ir.Status != ItemStatusAccepted {
+			t.Fatalf("expected item %+v to be accepted, got %s", ir.Item, ir.Status)
+		}
+	}
+}
+
+func TestProcess_rejectsInvalidItemsWithoutFailingWholeBatch(t *testing.T) {
+	s := NewService()
+	items := []Item{
+		{RecipientID: "rec_1", Amount: 100},
+		{RecipientID: "", Amount: 50},
+		{RecipientID: "rec_3", Amount: -5},
+	}
+
+	res, err := s.Process("idem_1", "bat_1", items)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res.Rejects != 2 {
+		t.Fatalf("expected 2 rejects, got %d", res.Rejects)
+	}
+	if res.Total != 100 {
+		t.Fatalf("expected total 100 (only the valid item), got %v", res.Total)
+	}
+}
+
+func TestProcess_rejectsEmptyBatch(t *testing.T) {
+	s := NewService()
+	_, err := s.Process("idem_1", "bat_1", nil)
+	if !errors.Is(err, ErrEmptyBatch) {
+		t.Fatalf("expected ErrEmptyBatch, got %v", err)
+	}
+}
+
+func TestProcess_rejectsTooManyItems(t *testing.T) {
+	s := NewService()
+	items := make([]Item, MaxItems+1)
+	for i := range items {
+		items[i] = Item{RecipientID: "rec", Amount: 1}
+	}
+
+	_, err := s.Process("idem_1", "bat_1", items)
+	if !errors.Is(err, ErrTooManyItems) {
+		t.Fatalf("expected ErrTooManyItems, got %v", err)
+	}
+}
+
+func TestProcess_isIdempotentForSameKeyAndBatch(t *testing.T) {
+	s := NewService()
+	items := []Item{{RecipientID: "rec_1", Amount: 100}}
+
+	first, err := s.Process("idem_1", "bat_1", items)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	second, err := s.Process("idem_1", "bat_1", items)
+	if err != nil {
+		t.Fatalf("unexpected error on replay: %v", err)
+	}
+	if second.ID != first.ID || second.Total != first.Total {
+		t.Fatalf("expected replay to return identical result, got %+v vs %+v", first, second)
+	}
+}
+
+func TestProcess_rejectsIdempotencyKeyReuseWithDifferentBatch(t *testing.T) {
+	s := NewService()
+	items := []Item{{RecipientID: "rec_1", Amount: 100}}
+
+	if _, err := s.Process("idem_1", "bat_1", items); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	_, err := s.Process("idem_1", "bat_2", items)
+	if !errors.Is(err, ErrDuplicateIdempotencyKey) {
+		t.Fatalf("expected ErrDuplicateIdempotencyKey, got %v", err)
+	}
+}