diff --git a/internal/batch/batch.go b/internal/batch/batch.go new file mode 100644 index 0000000..69e39fd --- /dev/null +++ b/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 +} diff --git a/internal/batch/batch_test.go b/internal/batch/batch_test.go new file mode 100644 index 0000000..d345099 --- /dev/null +++ b/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) + } +}