Merge pull request #1614 from adamdecaf/feat-merge-api

server: add Merge files endpoint

Author: adamdecaf

openapi.yaml

@@ -744,6 +744,41 @@ paths:
                 $ref: 'https://raw.githubusercontent.com/moov-io/base/master/api/common.yaml#/components/schemas/Error'
         '404':
           description: A resource with the specified ID was not found
+  /merge:
+    post:
+      tags: ['ACH Files']
+      summary: Merge Files
+      description: Combine multiple fileIDs and files together. Returned is the results of merging
+      operationId: mergeFiles
+      parameters:
+        - name: X-Request-ID
+          in: header
+          description: Optional Request ID allows application developer to trace requests through the system's logs
+          example: "rs4f9915"
+          schema:
+            type: string
+      requestBody:
+        description: JSON object containing FileIDs and ACH files (in JSON format)
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/MergeFilesRequest'
+      responses:
+        '200':
+          description: Files were merged together
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/MergeFilesResponse'
+        '400':
+          description: See error in response body
+          content:
+            application/json:
+              schema:
+                $ref: 'https://raw.githubusercontent.com/moov-io/base/master/api/common.yaml#/components/schemas/Error'
+        '404':
+          description: A resource with the specified ID was not found
 
 components:
   schemas:
@@ -2607,3 +2642,19 @@ components:
           $ref: '#/components/schemas/SegmentFileConfiguration'
         validateOpts:
           $ref: '#/components/schemas/ValidateOpts'
+    MergeFilesRequest:
+      properties:
+        fileIDs:
+          type: array
+          items:
+            type: string
+        files:
+          type: array
+          items:
+            $ref: '#/components/schemas/File'
+    MergeFilesResponse:
+      properties:
+        files:
+          type: array
+          items:
+            $ref: '#/components/schemas/File'

server/files.go

@@ -19,6 +19,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"encoding/json"
 	"errors"
@@ -821,3 +822,71 @@ func decodeFlattenBatchesRequest(_ context.Context, r *http.Request) (interface{
 		requestID: moovhttp.GetRequestID(r),
 	}, nil
 }
+
+type mergeFilesRequest struct {
+	FileIDs []string    `json:"fileIDs"`
+	Files   []*ach.File `json:"files"`
+
+	RequestID string `json:"requestID"`
+}
+
+type mergeFilesResponse struct {
+	Files []*ach.File `json:"files"`
+	Err   error       `json:"error"`
+}
+
+func mergeFilesEndpoint(s Service, r Repository, logger log.Logger) endpoint.Endpoint {
+	return func(_ context.Context, request interface{}) (interface{}, error) {
+		req, ok := request.(mergeFilesRequest)
+		if !ok {
+			return mergeFilesResponse{Err: ErrFoundABug}, ErrFoundABug
+		}
+
+		merged, err := s.MergeFiles(req.FileIDs, req.Files)
+		if logger != nil {
+			logger := logger.With(log.Fields{
+				"file_ids":  log.Strings(req.FileIDs),
+				"requestID": log.String(req.RequestID),
+			})
+			if err != nil {
+				logger.Error().LogError(err)
+			} else {
+				logger.Info().Log("merging files")
+			}
+		}
+		if err != nil {
+			return mergeFilesResponse{Err: err}, err
+		}
+
+		for idx := range merged {
+			err := r.StoreFile(merged[idx])
+			if logger != nil {
+				logger := logger.With(log.Fields{
+					"file_ids":  log.Strings(req.FileIDs),
+					"requestID": log.String(req.RequestID),
+				})
+				if err != nil {
+					logger.Error().LogError(err)
+				} else {
+					logger.Info().Logf("merge created file %s", merged[idx].ID)
+				}
+			}
+		}
+
+		return mergeFilesResponse{
+			Files: merged,
+		}, nil
+	}
+}
+
+func decodeMergeFilesRequest(_ context.Context, r *http.Request) (interface{}, error) {
+	var req mergeFilesRequest
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		return nil, fmt.Errorf("decoding merge files json: %w", err)
+	}
+
+	req.RequestID = cmp.Or(req.RequestID, moovhttp.GetRequestID(r))
+
+	return req, nil
+}

server/files_test.go

@@ -1442,3 +1442,55 @@ func TestFiles__flattenFileEndpointError(t *testing.T) {
 		t.Errorf("resp.Err=%q", resp.Err)
 	}
 }
+
+func TestFiles_MergeFiles(t *testing.T) {
+	logger := log.NewTestLogger()
+	repo := NewRepositoryInMemory(testTTLDuration, logger)
+	svc := NewService(repo)
+	router := MakeHTTPHandler(svc, repo, kitlog.NewLogfmtLogger(os.Stdout))
+
+	// Write a file into the repository
+	file, err := ach.ReadJSONFile(filepath.Join("..", "test", "testdata", "ppd-mixedDebitCredit-valid.json"))
+	require.NoError(t, err)
+
+	file.ID = base.ID()
+
+	err = repo.StoreFile(file)
+	require.NoError(t, err)
+
+	// Merge this with another file
+	file2, err := ach.ReadJSONFile(filepath.Join("..", "test", "testdata", "ppd-valid.json"))
+	require.NoError(t, err)
+
+	var body bytes.Buffer
+	err = json.NewEncoder(&body).Encode(mergeFilesRequest{
+		FileIDs: []string{file.ID},
+		Files:   []*ach.File{file2},
+	})
+	req := httptest.NewRequest("POST", "/merge", &body)
+
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+	w.Flush()
+
+	require.Equal(t, http.StatusOK, w.Code)
+
+	var resp mergeFilesResponse
+	err = json.NewDecoder(w.Body).Decode(&resp)
+	require.NoError(t, err)
+
+	require.Len(t, resp.Files, 1)
+
+	// Check the file
+	merged := resp.Files[0]
+	require.Len(t, merged.Batches, 2)
+
+	var foundEntries int
+	for idx := range merged.Batches {
+		foundEntries += len(merged.Batches[idx].GetEntries())
+	}
+	require.Equal(t, 3, foundEntries)
+
+	require.Equal(t, 100000, merged.Control.TotalDebitEntryDollarAmountInFile)
+	require.Equal(t, 200000, merged.Control.TotalCreditEntryDollarAmountInFile)
+}

server/routing.go

@@ -211,6 +211,12 @@ func MakeHTTPHandler(s Service, repo Repository, kitlog gokitlog.Logger) http.Ha
 		encodeResponse,
 		options...,
 	))
+	r.Methods("POST").Path("/merge").Handler(httptransport.NewServer(
+		mergeFilesEndpoint(s, repo, logger),
+		decodeMergeFilesRequest,
+		encodeResponse,
+		options...,
+	))
 	return r
 }
 

server/service.go

@@ -19,6 +19,8 @@ package server
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -65,6 +67,8 @@ type Service interface {
 	GetBatches(fileID string) []ach.Batcher
 	// DeleteBatch takes a fileID and BatchID and removes the batch from the file
 	DeleteBatch(fileID string, batchID string) error
+	// MergeFiles will combine all the given files together
+	MergeFiles(fileIDs []string, files []*ach.File) ([]*ach.File, error)
 }
 
 // service a concrete implementation of the service.
@@ -262,3 +266,37 @@ func (s *service) FlattenBatches(fileID string) (*ach.File, error) {
 	}
 	return ff, err
 }
+
+func (s *service) MergeFiles(fileIDs []string, files []*ach.File) ([]*ach.File, error) {
+	for idx := range fileIDs {
+		file, err := s.store.FindFile(fileIDs[idx])
+		if err != nil {
+			return nil, fmt.Errorf("file not found: %v", fileIDs[idx])
+		}
+
+		files = append(files, file)
+	}
+
+	merged, err := ach.MergeFiles(files)
+	if err != nil {
+		return nil, fmt.Errorf("merging files: %w", err)
+	}
+
+	for idx := range merged {
+		var buf bytes.Buffer
+		err := ach.NewWriter(&buf).Write(merged[idx])
+		if err != nil {
+			return nil, fmt.Errorf("problem hashing merged file: %w", err)
+		}
+
+		merged[idx].ID = hash(buf.Bytes())
+	}
+
+	return merged, nil
+}
+
+func hash(data []byte) string {
+	ss := sha256.New()
+	ss.Write(data)
+	return hex.EncodeToString(ss.Sum(nil))
+}