summercms/modules/tide/upstream.go

package tide

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"slices"
	"strings"
	"sync"

	"github.com/goccy/go-yaml"
)

// UpstreamSidecar is the upstream HTTP traffic a backend sent to outside
// services while one parity fixture was recorded. It lives next to the
// fixture as <fixture>.upstream.yaml (see UpstreamPath). Exchanges are
// consumed in recorded order across the whole flow.
type UpstreamSidecar struct {
	Version   int                `yaml:"version"`
	Exchanges []UpstreamExchange `yaml:"exchanges"`
}

// UpstreamExchange is one outbound request and the vendor's answer.
type UpstreamExchange struct {
	Request  UpstreamRequest  `yaml:"request"`
	Response UpstreamResponse `yaml:"response"`
}

// UpstreamRequest is the request the backend sent upstream. URL is absolute
// (scheme, host, path and query). Headers holds only the compared headers
// (UpstreamCompareHeaders); credential values are {{name}} placeholders.
type UpstreamRequest struct {
	Method  string            `yaml:"method"`
	URL     string            `yaml:"url"`
	Headers map[string]string `yaml:"headers,omitempty"`
	Body    string            `yaml:"body,omitempty"`
}

// UpstreamResponse is the recorded vendor answer the fake replays.
type UpstreamResponse struct {
	Status  int               `yaml:"status"`
	Headers map[string]string `yaml:"headers,omitempty"`
	Body    string            `yaml:"body,omitempty"`
}

// UpstreamCompareHeaders are the request headers the fake asserts. A header
// is compared when either the recorded or the sent request carries it.
var UpstreamCompareHeaders = []string{
	"User-Agent",
	"Accept",
	"Content-Type",
	"Authorization",
	"X-Api-Key",
	"Anthropic-Version",
	"Anthropic-Beta",
}

// UpstreamPath maps a fixture path to its sidecar path: x.yaml becomes
// x.upstream.yaml.
func UpstreamPath(fixturePath string) string {
	for _, ext := range []string{".yaml", ".yml"} {
		if strings.HasSuffix(fixturePath, ext) {
			return strings.TrimSuffix(fixturePath, ext) + ".upstream.yaml"
		}
	}
	return fixturePath + ".upstream.yaml"
}

// LoadUpstream reads a version-1 sidecar, rejecting unknown fields. A missing
// file returns an error wrapping fs.ErrNotExist.
func LoadUpstream(path string) (UpstreamSidecar, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return UpstreamSidecar{}, fmt.Errorf("tide: read upstream %s: %w", path, err)
	}
	var s UpstreamSidecar
	dec := yaml.NewDecoder(bytes.NewReader(raw), yaml.DisallowUnknownField())
	if err := dec.Decode(&s); err != nil {
		return UpstreamSidecar{}, fmt.Errorf("tide: parse upstream %s: %w", path, err)
	}
	if err := validateUpstream(s); err != nil {
		return UpstreamSidecar{}, fmt.Errorf("tide: upstream %s: %w", path, err)
	}
	return s, nil
}

func validateUpstream(s UpstreamSidecar) error {
	if s.Version != CurrentVersion {
		return fmt.Errorf("version %d, want %d", s.Version, CurrentVersion)
	}
	for i, ex := range s.Exchanges {
		if strings.TrimSpace(ex.Request.Method) == "" {
			return fmt.Errorf("exchange %d: request method is required", i)
		}
		u, err := url.Parse(ex.Request.URL)
		if err != nil || u.Scheme == "" || u.Host == "" {
			return fmt.Errorf("exchange %d: request url %q must be absolute", i, ex.Request.URL)
		}
		if ex.Response.Status < 100 || ex.Response.Status > 999 {
			return fmt.Errorf("exchange %d: response status %d out of range", i, ex.Response.Status)
		}
	}
	return nil
}

// UpstreamFake is an http.RoundTripper that answers from a sidecar without
// dialing. Each request takes the next recorded exchange and is asserted
// against it: method, scheme, host, path, query (order-insensitive), the
// UpstreamCompareHeaders with {{name}} placeholders expanded from the store,
// and the body (JSON semantically, otherwise byte for byte). A mismatch fails
// the request and is remembered for Verify.
type UpstreamFake struct {
	mu        sync.Mutex
	store     *Store
	exchanges []UpstreamExchange
	next      int
	errs      []error
}

// NewUpstreamFake returns a fake that replays s. store resolves {{name}}
// placeholders in the recorded requests and responses; it may be nil when the
// sidecar has none.
func NewUpstreamFake(s UpstreamSidecar, store *Store) *UpstreamFake {
	return &UpstreamFake{store: store, exchanges: slices.Clone(s.Exchanges)}
}

// RoundTrip implements http.RoundTripper.
func (f *UpstreamFake) RoundTrip(req *http.Request) (*http.Response, error) {
	var body []byte
	if req.Body != nil {
		var err error
		body, err = io.ReadAll(io.LimitReader(req.Body, DefaultMaxBody+1))
		_ = req.Body.Close()
		if err != nil {
			return nil, f.fail(fmt.Errorf("tide: upstream %s %s: read body: %w", req.Method, req.URL, err))
		}
	}
	f.mu.Lock()
	if f.next >= len(f.exchanges) {
		f.mu.Unlock()
		return nil, f.fail(fmt.Errorf("tide: upstream extra request %s %s", req.Method, redactURL(req.URL)))
	}
	idx := f.next
	ex := f.exchanges[idx]
	f.next++
	f.mu.Unlock()

	if problems := f.compare(ex.Request, req, body); len(problems) > 0 {
		return nil, f.fail(fmt.Errorf("tide: upstream exchange %d %s %s: %s",
			idx, req.Method, redactURL(req.URL), strings.Join(problems, "; ")))
	}
	resp, err := f.response(ex.Response, req)
	if err != nil {
		return nil, f.fail(fmt.Errorf("tide: upstream exchange %d: %w", idx, err))
	}
	return resp, nil
}

// Verify reports every mismatch, every extra request and every recorded
// exchange that was never requested. It returns nil when the backend sent
// exactly the recorded requests.
func (f *UpstreamFake) Verify() error {
	f.mu.Lock()
	defer f.mu.Unlock()
	errs := slices.Clone(f.errs)
	for i := f.next; i < len(f.exchanges); i++ {
		r := f.exchanges[i].Request
		errs = append(errs, fmt.Errorf("tide: upstream exchange %d unconsumed: %s %s", i, r.Method, r.URL))
	}
	return errors.Join(errs...)
}

func (f *UpstreamFake) fail(err error) error {
	f.mu.Lock()
	f.errs = append(f.errs, err)
	f.mu.Unlock()
	return err
}

func (f *UpstreamFake) compare(want UpstreamRequest, got *http.Request, body []byte) []string {
	var problems []string
	if !strings.EqualFold(want.Method, got.Method) {
		problems = append(problems, fmt.Sprintf("method: want %s, got %s", want.Method, got.Method))
	}
	rawURL, err := f.store.Expand(want.URL)
	if err != nil {
		return append(problems, "url: "+err.Error())
	}
	wu, err := url.Parse(rawURL)
	if err != nil {
		return append(problems, "url: "+err.Error())
	}
	gu := got.URL
	if !strings.EqualFold(wu.Scheme, gu.Scheme) {
		problems = append(problems, fmt.Sprintf("scheme: want %s, got %s", wu.Scheme, gu.Scheme))
	}
	if !strings.EqualFold(wu.Host, gu.Host) {
		problems = append(problems, fmt.Sprintf("host: want %s, got %s", wu.Host, gu.Host))
	}
	if wu.EscapedPath() != gu.EscapedPath() && wu.Path != gu.Path {
		problems = append(problems, fmt.Sprintf("path: want %s, got %s", wu.EscapedPath(), gu.EscapedPath()))
	}
	problems = append(problems, compareQuery(wu.Query(), gu.Query())...)

	for _, name := range UpstreamCompareHeaders {
		wv, err := f.store.Expand(headerValue(want.Headers, name))
		if err != nil {
			problems = append(problems, "header "+name+": "+err.Error())
			continue
		}
		gv := got.Header.Get(name)
		if wv == gv {
			continue
		}
		if isCredentialHeader(name) {
			problems = append(problems, fmt.Sprintf("header %s: value differs", name))
			continue
		}
		problems = append(problems, fmt.Sprintf("header %s: want %q, got %q", name, wv, gv))
	}

	wantBody, err := f.store.Expand(want.Body)
	if err != nil {
		return append(problems, "body: "+err.Error())
	}
	jsonBody := isJSONContentType(want.Headers) || isJSONContentType(map[string]string{"Content-Type": got.Header.Get("Content-Type")})
	switch {
	case jsonBody && (len(wantBody) > 0 || len(body) > 0):
		if diffs := diffJSON([]byte(wantBody), body); len(diffs) > 0 {
			d := diffs[0]
			problems = append(problems, fmt.Sprintf("body %s: want %s, got %s", d.Path, d.Expected, d.Actual))
		}
	case wantBody != string(body):
		problems = append(problems, fmt.Sprintf("body: want %d bytes, got %d bytes", len(wantBody), len(body)))
	}
	return problems
}

func compareQuery(want, got url.Values) []string {
	var problems []string
	keys := make([]string, 0, len(want)+len(got))
	for k := range want {
		keys = append(keys, k)
	}
	for k := range got {
		if _, ok := want[k]; !ok {
			keys = append(keys, k)
		}
	}
	slices.Sort(keys)
	for _, k := range keys {
		wv, gv := slices.Clone(want[k]), slices.Clone(got[k])
		slices.Sort(wv)
		slices.Sort(gv)
		if !slices.Equal(wv, gv) {
			problems = append(problems, fmt.Sprintf("query %s: want %q, got %q", k, wv, gv))
		}
	}
	return problems
}

func (f *UpstreamFake) response(r UpstreamResponse, req *http.Request) (*http.Response, error) {
	h := http.Header{}
	for k, v := range r.Headers {
		ev, err := f.store.Expand(v)
		if err != nil {
			return nil, fmt.Errorf("response header %s: %w", k, err)
		}
		h.Set(k, ev)
	}
	body, err := f.store.Expand(r.Body)
	if err != nil {
		return nil, fmt.Errorf("response body: %w", err)
	}
	return &http.Response{
		Status:        fmt.Sprintf("%d %s", r.Status, http.StatusText(r.Status)),
		StatusCode:    r.Status,
		Proto:         "HTTP/1.1",
		ProtoMajor:    1,
		ProtoMinor:    1,
		Header:        h,
		Body:          io.NopCloser(strings.NewReader(body)),
		ContentLength: int64(len(body)),
		Request:       req,
	}, nil
}

func isCredentialHeader(name string) bool {
	return strings.EqualFold(name, "Authorization") || strings.EqualFold(name, "X-Api-Key")
}

// redactURL drops the query, which may carry a vendor token.
func redactURL(u *url.URL) string {
	if u == nil {
		return ""
	}
	c := *u
	c.RawQuery = ""
	c.User = nil
	return c.String()
}