summercms/modules/tide/centrifugo.go

package tide

import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"
)

// DefaultCentrifugoListen is the loopback address the fake Centrifugo
// recorder binds when no other address is given.
const DefaultCentrifugoListen = "127.0.0.1:8424"

// MaxPublicationBody caps one recorded Centrifugo API request body.
const MaxPublicationBody = 1 << 20

// DefaultBroadcastSettle is how long RecordBroadcasts waits after a step
// for publications that arrive after its HTTP response.
const DefaultBroadcastSettle = 500 * time.Millisecond

// Publication is one publish or broadcast request received by the fake
// Centrifugo recorder. The Authorization header value is never kept: only
// whether it carried the expected API key.
type Publication struct {
	Method        string          `yaml:"method"`
	Path          string          `yaml:"path"`
	Authorization bool            `yaml:"authorization"`
	Body          json.RawMessage `yaml:"body"`
}

// CentrifugoRecorderOptions configures NewCentrifugoRecorder.
type CentrifugoRecorderOptions struct {
	// APIKey is the key the backend under test sends as
	// "Authorization: apikey <key>". It is only compared, never stored.
	APIKey string
}

// CentrifugoRecorder is a fake Centrifugo HTTP API. It records every
// publish and broadcast request and answers the other server API calls a
// backend makes with empty successful results.
type CentrifugoRecorder struct {
	apiKey string
	mu     sync.Mutex
	pubs   []Publication
}

// NewCentrifugoRecorder returns an empty recorder.
func NewCentrifugoRecorder(opts CentrifugoRecorderOptions) *CentrifugoRecorder {
	return &CentrifugoRecorder{apiKey: opts.APIKey}
}

// ServeHTTP records POST .../publish and .../broadcast and answers
// {"result":{}}. .../presence answers an empty presence map, .../unsubscribe
// and .../info answer {"result":{}}, anything else is 404.
func (c *CentrifugoRecorder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	path := strings.TrimRight(r.URL.Path, "/")
	recorded := strings.HasSuffix(path, "/publish") || strings.HasSuffix(path, "/broadcast")
	answered := strings.HasSuffix(path, "/presence") || strings.HasSuffix(path, "/unsubscribe") || strings.HasSuffix(path, "/info")
	if !recorded && !answered {
		http.NotFound(w, r)
		return
	}
	if r.Method != http.MethodPost {
		w.Header().Set("Allow", http.MethodPost)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	raw, err := io.ReadAll(io.LimitReader(r.Body, MaxPublicationBody+1))
	if err != nil {
		http.Error(w, "read body", http.StatusBadRequest)
		return
	}
	if len(raw) > MaxPublicationBody {
		http.Error(w, "body too large", http.StatusRequestEntityTooLarge)
		return
	}
	if recorded {
		c.mu.Lock()
		c.pubs = append(c.pubs, Publication{
			Method:        r.Method,
			Path:          r.URL.Path,
			Authorization: c.authorized(r.Header.Get("Authorization")),
			Body:          json.RawMessage(raw),
		})
		c.mu.Unlock()
	}
	w.Header().Set("Content-Type", "application/json")
	if strings.HasSuffix(path, "/presence") {
		_, _ = io.WriteString(w, `{"result":{"presence":{}}}`)
		return
	}
	_, _ = io.WriteString(w, `{"result":{}}`)
}

func (c *CentrifugoRecorder) authorized(header string) bool {
	if c.apiKey == "" {
		return false
	}
	want := "apikey " + c.apiKey
	return subtle.ConstantTimeCompare([]byte(header), []byte(want)) == 1
}

// Publications returns a copy of the recorded requests in arrival order.
func (c *CentrifugoRecorder) Publications() []Publication {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := make([]Publication, len(c.pubs))
	for i, p := range c.pubs {
		p.Body = append(json.RawMessage(nil), p.Body...)
		out[i] = p
	}
	return out
}

// Reset forgets every recorded request.
func (c *CentrifugoRecorder) Reset() {
	c.mu.Lock()
	c.pubs = nil
	c.mu.Unlock()
}

// ListenAndServe serves the recorder on addr until ctx is cancelled. addr
// must be a loopback address, the same rule the recording proxy applies.
func (c *CentrifugoRecorder) ListenAndServe(ctx context.Context, addr string) error {
	if err := requireLoopbackAddr(addr); err != nil {
		return err
	}
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("tide: centrifugo recorder listen %s: %w", addr, err)
	}
	srv := &http.Server{Handler: c, ReadHeaderTimeout: 10 * time.Second}
	done := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
			shutCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			_ = srv.Shutdown(shutCtx)
			cancel()
		case <-done:
		}
	}()
	err = srv.Serve(ln)
	close(done)
	if errors.Is(err, http.ErrServerClosed) {
		return nil
	}
	return err
}

// BroadcastConfig drives RecordBroadcasts.
type BroadcastConfig struct {
	// Target is the loopback base URL of the reference backend.
	Target string
	// Listen is the loopback address of the recorder the backend publishes
	// to. Empty means DefaultCentrifugoListen.
	Listen string
	// APIKey is the Centrifugo API key the backend is configured with.
	APIKey string
	// Store holds and receives captured variables (the private vars file).
	Store *Store
	// Rules are optional capture rules merged into steps without captures.
	Rules Rules
	// Step names the flow step whose publications are recorded. Earlier
	// steps run first as setup and their publications are discarded; later
	// steps do not run. Empty records the whole flow.
	Step string
	// IDs names the id:* variables replaced by placeholders. Empty means
	// every id:* placeholder or capture the flow mentions.
	IDs []string
	// Name is the golden name. Empty uses the step, else the flow name.
	Name string
	// Settle is the wait for publications that arrive after a response.
	Settle time.Duration
	// Client and MaxBody are passed to RecordFlow.
	Client  *http.Client
	MaxBody int64
}

// RecordBroadcasts runs spec against the reference backend while a
// loopback CentrifugoRecorder stands in for Centrifugo, then returns the
// normalised publications as a golden. It fails when nothing is published.
func RecordBroadcasts(ctx context.Context, spec Flow, cfg BroadcastConfig) (BroadcastGolden, error) {
	if err := validateFlow(spec); err != nil {
		return BroadcastGolden{}, err
	}
	if _, err := parseLoopbackUpstream(cfg.Target); err != nil {
		return BroadcastGolden{}, fmt.Errorf("tide: broadcast target: %w", err)
	}
	if strings.TrimSpace(cfg.APIKey) == "" {
		return BroadcastGolden{}, fmt.Errorf("tide: broadcast recording needs the backend's Centrifugo API key")
	}
	listen := cfg.Listen
	if listen == "" {
		listen = DefaultCentrifugoListen
	}
	if err := requireLoopbackAddr(listen); err != nil {
		return BroadcastGolden{}, err
	}
	idx := -1
	if cfg.Step != "" {
		for i, s := range spec.Steps {
			if s.ID == cfg.Step {
				idx = i
				break
			}
		}
		if idx < 0 {
			return BroadcastGolden{}, fmt.Errorf("tide: flow %q has no step %q", spec.Name, cfg.Step)
		}
	}
	store := cfg.Store
	if store == nil {
		store = mustMemoryStore()
	}
	settle := cfg.Settle
	if settle <= 0 {
		settle = DefaultBroadcastSettle
	}

	rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: cfg.APIKey})
	srvCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	errCh := make(chan error, 1)
	go func() { errCh <- rec.ListenAndServe(srvCtx, listen) }()
	if err := waitListening(ctx, listen, errCh); err != nil {
		return BroadcastGolden{}, err
	}

	record := func(steps []Step) error {
		sub := spec
		sub.Steps = steps
		_, err := RecordFlow(ctx, sub, RecordConfig{Target: cfg.Target, Client: cfg.Client, MaxBody: cfg.MaxBody, Store: store, Rules: cfg.Rules})
		if err != nil {
			return err
		}
		return sleepCtx(ctx, settle)
	}
	steps := spec.Steps
	if idx >= 0 {
		if idx > 0 {
			if err := record(spec.Steps[:idx]); err != nil {
				return BroadcastGolden{}, err
			}
			rec.Reset()
		}
		steps = spec.Steps[idx : idx+1]
	}
	if err := record(steps); err != nil {
		return BroadcastGolden{}, err
	}
	pubs := rec.Publications()
	if len(pubs) == 0 {
		return BroadcastGolden{}, fmt.Errorf("tide: flow %q published nothing to %s; is the backend's Centrifugo API URL pointed at the recorder?", spec.Name, listen)
	}

	names := cfg.IDs
	if len(names) == 0 {
		names = flowIDNames(spec)
	}
	idStore := mustMemoryStore()
	for _, n := range names {
		v, ok := store.Get(n)
		if !ok || v == "" {
			return BroadcastGolden{}, fmt.Errorf("tide: id variable %q is not in the vars store", n)
		}
		idStore.Set(n, v)
	}
	norm, err := NormalizePublications(pubs, idStore)
	if err != nil {
		return BroadcastGolden{}, err
	}
	for i, p := range norm {
		if strings.Contains(string(p.Body), cfg.APIKey) {
			return BroadcastGolden{}, fmt.Errorf("tide: publication %d body contains the Centrifugo API key", i)
		}
	}
	g := BroadcastGolden{
		Version:      CurrentVersion,
		Name:         cfg.Name,
		Flow:         spec.Name,
		Publications: norm,
	}
	if cfg.Step != "" {
		g.Flow += "#" + cfg.Step
	}
	if g.Name == "" {
		g.Name = cfg.Step
	}
	if g.Name == "" {
		g.Name = spec.Name
	}
	return g, nil
}

func waitListening(ctx context.Context, addr string, errCh <-chan error) error {
	deadline := time.Now().Add(5 * time.Second)
	for {
		select {
		case err := <-errCh:
			if err == nil {
				err = fmt.Errorf("stopped")
			}
			return fmt.Errorf("tide: centrifugo recorder: %w", err)
		default:
		}
		conn, err := net.DialTimeout("tcp", addr, 200*time.Millisecond)
		if err == nil {
			_ = conn.Close()
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("tide: centrifugo recorder did not start on %s: %w", addr, err)
		}
		if err := sleepCtx(ctx, 20*time.Millisecond); err != nil {
			return err
		}
	}
}

func sleepCtx(ctx context.Context, d time.Duration) error {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-t.C:
		return nil
	}
}

// flowIDNames lists the id:* variables a flow mentions in request
// placeholders or captures, sorted.
func flowIDNames(spec Flow) []string {
	seen := map[string]struct{}{}
	add := func(s string) {
		for _, m := range placeholderRe.FindAllStringSubmatch(s, -1) {
			if strings.HasPrefix(m[1], "id:") {
				seen[m[1]] = struct{}{}
			}
		}
	}
	for _, step := range spec.Steps {
		add(step.Request.Path)
		add(step.Request.Query)
		add(string(step.Request.Body))
		for _, v := range step.Request.Headers {
			add(v)
		}
		for _, c := range step.Capture {
			if strings.HasPrefix(c.As, "id:") {
				seen[c.As] = struct{}{}
			}
		}
	}
	out := make([]string, 0, len(seen))
	for n := range seen {
		out = append(out, n)
	}
	sort.Strings(out)
	return out
}