payments/ee/plugins/routable/client/client.go

package client

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/hanzo-fi/payments/pkg/domain/httpwrapper"
	"github.com/hanzo-fi/payments/pkg/domain/metrics"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

// DefaultBaseURL is the production Routable API endpoint. Sandbox is
// https://api.sandbox.routable.com (selected via the connector config).
const DefaultBaseURL = "https://api.routable.com"

//go:generate mockgen -source client.go -destination client_generated.go -package client . Client
type Client interface {
	ListAccounts(ctx context.Context, page, pageSize int) (*ListAccountsResponse, error)
	GetAccount(ctx context.Context, id string) (*Account, error)

	ListCompanies(ctx context.Context, page, pageSize int) (*ListCompaniesResponse, error)

	ListPayables(ctx context.Context, page, pageSize int, statusChangedAtGte time.Time) (*ListPayablesResponse, error)
	GetPayable(ctx context.Context, id string) (*Payable, error)
	// CreatePayable returns the parsed Payable, the HTTP status code, and
	// any error. Callers branch on the status code: 201 means Routable
	// echoed the full payable model (sync); 202 means the payable was
	// accepted but only `{id, status}` was returned and the caller must
	// poll. Mapping a 202 payable as if it were complete throws away
	// information and surfaces as a misleading "unsupported currency"
	// error downstream — surface the status here instead.
	CreatePayable(ctx context.Context, req CreatePayableRequest) (*Payable, int, error)

	ListReceivables(ctx context.Context, page, pageSize int, statusChangedAtGte time.Time) (*ListReceivablesResponse, error)
}

type client struct {
	httpClient httpwrapper.Client
	endpoint   string
}

type apiTransport struct {
	apiKey     string
	underlying http.RoundTripper
}

func (t *apiTransport) RoundTrip(req *http.Request) (*http.Response, error) {
	req.Header.Set("Authorization", "Bearer "+t.apiKey)
	req.Header.Set("Accept", "application/json")
	if req.Header.Get("Content-Type") == "" {
		req.Header.Set("Content-Type", "application/json")
	}
	return t.underlying.RoundTrip(req)
}

// New wires the Routable client through Formance's standard HTTP layer
// (otel + metrics + error mapping). connectorName is the metrics label.
func New(connectorName, apiKey, endpoint string) Client {
	endpoint = strings.TrimSuffix(endpoint, "/")
	if endpoint == "" {
		endpoint = DefaultBaseURL
	}

	transport := &apiTransport{
		apiKey:     apiKey,
		underlying: otelhttp.NewTransport(tunedHTTPTransport()),
	}

	return &client{
		endpoint: endpoint,
		httpClient: httpwrapper.NewClient(&httpwrapper.Config{
			Transport: metrics.NewTransport(connectorName, metrics.TransportOpts{Transport: transport}),
		}),
	}
}

// tunedHTTPTransport keeps the connection pool warm at sustained
// 100-200 req/min. Default MaxIdleConnsPerHost=2 churns TCP/TLS
// handshakes at 200k tx/wk; 64 keeps reuse high without hogging the host.
func tunedHTTPTransport() *http.Transport {
	t := http.DefaultTransport.(*http.Transport).Clone()
	t.MaxIdleConnsPerHost = 64
	t.MaxConnsPerHost = 0 // unbounded — engine concurrency caps the upper bound
	t.IdleConnTimeout = 90 * time.Second
	return t
}

func (c *client) buildURL(path string, query url.Values) string {
	u := c.endpoint + path
	if len(query) == 0 {
		return u
	}
	return u + "?" + query.Encode()
}

func paginationQuery(page, pageSize int) url.Values {
	q := url.Values{}
	if page > 0 {
		q.Set("page", strconv.Itoa(page))
	}
	if pageSize > 0 {
		q.Set("page_size", strconv.Itoa(pageSize))
	}
	return q
}

// do unmarshals the success body or the Routable error envelope.
func (c *client) do(ctx context.Context, method, path string, query url.Values, body any, idempotencyKey string, out any) (int, error) {
	reqBody := io.Reader(http.NoBody)
	if body != nil {
		buf := bytes.NewBuffer(nil)
		if err := json.NewEncoder(buf).Encode(body); err != nil {
			return 0, fmt.Errorf("encoding request body: %w", err)
		}
		reqBody = buf
	}

	req, err := http.NewRequestWithContext(ctx, method, c.buildURL(path, query), reqBody)
	if err != nil {
		return 0, fmt.Errorf("creating request: %w", err)
	}
	if idempotencyKey != "" {
		req.Header.Set("Idempotency-Key", idempotencyKey)
	}

	var apiErr ErrorResponse
	statusCode, doErr := c.httpClient.Do(ctx, req, out, &apiErr)
	if doErr == nil {
		return statusCode, nil
	}

	// Attach the Routable error envelope when the response carried one.
	// Transport errors (DNS, timeout, connection refused) leave apiErr
	// at its zero value, in which case appending the formatted envelope
	// just produces a misleading "empty body" suffix.
	if apiErr.hasContent() {
		return statusCode, fmt.Errorf("%w: %s", doErr, apiErr.Error())
	}
	return statusCode, doErr
}

func (c *client) ListAccounts(ctx context.Context, page, pageSize int) (*ListAccountsResponse, error) {
	var resp ListAccountsResponse
	if _, err := c.do(ctx, http.MethodGet, "/v1/settings/accounts", paginationQuery(page, pageSize), nil, "", &resp); err != nil {
		return nil, fmt.Errorf("listing accounts: %w", err)
	}
	return &resp, nil
}

func (c *client) GetAccount(ctx context.Context, id string) (*Account, error) {
	if id == "" {
		return nil, errors.New("account id is required")
	}
	var resp Account
	if _, err := c.do(ctx, http.MethodGet, "/v1/settings/accounts/"+url.PathEscape(id), nil, nil, "", &resp); err != nil {
		return nil, fmt.Errorf("getting account %s: %w", id, err)
	}
	return &resp, nil
}

func (c *client) ListCompanies(ctx context.Context, page, pageSize int) (*ListCompaniesResponse, error) {
	var resp ListCompaniesResponse
	if _, err := c.do(ctx, http.MethodGet, "/v1/companies", paginationQuery(page, pageSize), nil, "", &resp); err != nil {
		return nil, fmt.Errorf("listing companies: %w", err)
	}
	return &resp, nil
}

func (c *client) ListPayables(ctx context.Context, page, pageSize int, statusChangedAtGte time.Time) (*ListPayablesResponse, error) {
	q := paginationQuery(page, pageSize)
	if !statusChangedAtGte.IsZero() {
		q.Set("status_changed_at.gte", statusChangedAtGte.UTC().Format(time.RFC3339))
	}
	var resp ListPayablesResponse
	if _, err := c.do(ctx, http.MethodGet, "/v1/payables", q, nil, "", &resp); err != nil {
		return nil, fmt.Errorf("listing payables: %w", err)
	}
	return &resp, nil
}

func (c *client) GetPayable(ctx context.Context, id string) (*Payable, error) {
	if id == "" {
		return nil, errors.New("payable id is required")
	}
	var resp Payable
	statusCode, err := c.do(ctx, http.MethodGet, "/v1/payables/"+url.PathEscape(id), nil, nil, "", &resp)
	if err != nil {
		if statusCode == http.StatusNotFound {
			return nil, ErrPayableNotFound
		}
		return nil, fmt.Errorf("getting payable %s: %w", id, err)
	}
	return &resp, nil
}

func (c *client) CreatePayable(ctx context.Context, req CreatePayableRequest) (*Payable, int, error) {
	if err := validateCreatePayable(req); err != nil {
		return nil, 0, err
	}
	var resp Payable
	status, err := c.do(ctx, http.MethodPost, "/v1/payables", nil, req, req.IdempotencyKey, &resp)
	if err != nil {
		return nil, status, fmt.Errorf("creating payable: %w", err)
	}
	return &resp, status, nil
}

func validateCreatePayable(req CreatePayableRequest) error {
	switch {
	case req.Type == "":
		return errors.New("create payable: type is required")
	case req.DeliveryMethod == "":
		return errors.New("create payable: delivery_method is required")
	case req.PayToCompany == "":
		return errors.New("create payable: pay_to_company is required")
	case req.WithdrawFromAccount == "":
		return errors.New("create payable: withdraw_from_account is required")
	case req.Amount == "":
		return errors.New("create payable: amount is required")
	case len(req.LineItems) == 0:
		return errors.New("create payable: at least one line item is required")
	case req.ActingTeamMember == "":
		return errors.New("create payable: acting_team_member is required")
	}
	return nil
}

func (c *client) ListReceivables(ctx context.Context, page, pageSize int, statusChangedAtGte time.Time) (*ListReceivablesResponse, error) {
	q := paginationQuery(page, pageSize)
	if !statusChangedAtGte.IsZero() {
		q.Set("status_changed_at.gte", statusChangedAtGte.UTC().Format(time.RFC3339))
	}
	var resp ListReceivablesResponse
	if _, err := c.do(ctx, http.MethodGet, "/v1/receivables", q, nil, "", &resp); err != nil {
		return nil, fmt.Errorf("listing receivables: %w", err)
	}
	return &resp, nil
}