pkg-proxy/internal/server/change_feeds_test.go

package server

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/git-pkgs/proxy/internal/config"
	"github.com/git-pkgs/proxy/internal/database"
	"github.com/git-pkgs/proxy/internal/handler"
	"github.com/git-pkgs/proxy/internal/storage"
	"github.com/git-pkgs/registries/changes"
	"github.com/git-pkgs/registries/client"
	"github.com/git-pkgs/registries/fetch"
	"github.com/go-chi/chi/v5"
)

// changingRegistry serves npm packuments, Composer metadata, and both
// registries' change feeds, all from one set of documents that a test can
// change. The feeds list every document changed since the previous read.
type changingRegistry struct {
	*httptest.Server
	mu      sync.Mutex
	docs    map[string]string // path -> body
	changed []string          // paths changed since the feed was last read
	feedSeq int
	fetches map[string]int
}

func newChangingRegistry(t *testing.T) *changingRegistry {
	t.Helper()
	reg := &changingRegistry{docs: map[string]string{}, fetches: map[string]int{}}
	reg.Server = httptest.NewServer(http.HandlerFunc(reg.serve))
	t.Cleanup(reg.Close)
	return reg
}

func (reg *changingRegistry) set(path, body string) {
	reg.mu.Lock()
	defer reg.mu.Unlock()
	if _, ok := reg.docs[path]; ok {
		reg.changed = append(reg.changed, path)
	}
	reg.docs[path] = body
}

func (reg *changingRegistry) fetched(path string) int {
	reg.mu.Lock()
	defer reg.mu.Unlock()
	return reg.fetches[path]
}

// takeChanged returns the changed paths with prefix and forgets them.
func (reg *changingRegistry) takeChanged(prefix string) []string {
	var taken, kept []string
	for _, p := range reg.changed {
		if name, ok := strings.CutPrefix(p, prefix); ok {
			taken = append(taken, name)
		} else {
			kept = append(kept, p)
		}
	}
	reg.changed = kept
	return taken
}

func (reg *changingRegistry) serve(w http.ResponseWriter, r *http.Request) {
	reg.mu.Lock()
	defer reg.mu.Unlock()
	w.Header().Set("Content-Type", "application/json")
	switch r.URL.Path {
	case "/metadata/changes.json":
		if r.URL.Query().Get("since") == "" {
			w.WriteHeader(http.StatusBadRequest)
			_, _ = io.WriteString(w, `{"error":"missing since","timestamp":1}`)
			return
		}
		var actions []string
		for _, name := range reg.takeChanged("/p2/") {
			actions = append(actions, fmt.Sprintf(`{"type":"update","package":%q,"time":1}`, strings.TrimSuffix(name, ".json")))
		}
		_, _ = fmt.Fprintf(w, `{"actions":[%s],"timestamp":2}`, strings.Join(actions, ","))
		return
	case "/replicate/registry/":
		_, _ = fmt.Fprintf(w, `{"update_seq":%d}`, reg.feedSeq)
		return
	case "/replicate/registry/_changes":
		var results []string
		for _, name := range reg.takeChanged("/npm/") {
			reg.feedSeq++
			results = append(results, fmt.Sprintf(`{"seq":%d,"id":%q}`, reg.feedSeq, name))
		}
		_, _ = fmt.Fprintf(w, `{"results":[%s],"last_seq":%d}`, strings.Join(results, ","), reg.feedSeq)
		return
	}
	reg.fetches[r.URL.Path]++
	body, ok := reg.docs[r.URL.Path]
	if !ok {
		http.NotFound(w, r)
		return
	}
	etag := fmt.Sprintf(`"%x"`, sha256.Sum256([]byte(body)))
	if r.Header.Get("If-None-Match") == etag {
		w.WriteHeader(http.StatusNotModified)
		return
	}
	w.Header().Set("ETag", etag)
	_, _ = io.WriteString(w, body)
}

func composerDoc(versions ...string) string {
	var entries []string
	for _, v := range versions {
		entries = append(entries, fmt.Sprintf(`{"name":"acme/lib","version":%q,"version_normalized":"%s.0","dist":{"type":"zip","url":"https://example.test/acme/lib/%s.zip","reference":"r%s"}}`, v, v, v, v))
	}
	return `{"packages":{"acme/lib":[` + strings.Join(entries, ",") + `]},"minified":"composer/2.0"}`
}

func npmDoc(versions ...string) string {
	var entries []string
	for _, v := range versions {
		entries = append(entries, fmt.Sprintf(`%q:{"name":"left-pad","version":%q,"dist":{"tarball":"https://registry.example.test/left-pad/-/left-pad-%s.tgz"}}`, v, v, v))
	}
	return `{"name":"left-pad","dist-tags":{"latest":"` + versions[len(versions)-1] + `"},"versions":{` + strings.Join(entries, ",") + `}}`
}

// changeFeedRouter mounts the npm and Composer handlers on reg, with
// metadata cached for an hour.
func changeFeedRouter(t *testing.T, reg *changingRegistry) (http.Handler, *database.DB) {
	t.Helper()
	db, err := database.Create(t.TempDir() + "/test.db")
	if err != nil {
		t.Fatalf("creating database: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	store, err := storage.OpenBucket(context.Background(), "file://"+t.TempDir())
	if err != nil {
		t.Fatalf("opening storage: %v", err)
	}
	fetcher := fetch.NewFetcher(fetch.WithHTTPClient(reg.Client()), fetch.WithMaxRetries(0))
	t.Cleanup(func() { _ = fetcher.Close() })
	proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), slog.New(slog.NewTextHandler(io.Discard, nil)))
	proxy.HTTPClient = reg.Client()
	proxy.CacheMetadata = true
	proxy.MetadataTTL = time.Hour
	proxy.SetMetadataRewriteCacheSize(1 << 20)

	cfg := config.Default()
	cfg.BaseURL = "http://proxy.example"
	cfg.Upstream.NPM = reg.URL + "/npm"
	cfg.Upstream.Composer = reg.URL
	cfg.Upstream.ComposerRepository = reg.URL
	s := &Server{cfg: cfg}
	r := chi.NewRouter()
	s.mountProtocolHandlers(r, proxy)
	return r, db
}

func getBody(t *testing.T, h http.Handler, path string) string {
	t.Helper()
	w := httptest.NewRecorder()
	h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil))
	if w.Code != http.StatusOK {
		t.Fatalf("GET %s = %d: %s", path, w.Code, w.Body.String())
	}
	return w.Body.String()
}

// TestChangeFeedRevalidatesChangedMetadata checks the whole path: cached
// metadata is served within its TTL even after upstream changed, until the
// change feed lists the package, after which the next request revalidates
// and serves the new release. Packages the feed doesn't list stay cached.
func TestChangeFeedRevalidatesChangedMetadata(t *testing.T) {
	reg := newChangingRegistry(t)
	reg.set("/p2/acme/lib.json", composerDoc("1.0.0"))
	reg.set("/p2/acme/other.json", `{"packages":{"acme/other":[]},"minified":"composer/2.0"}`)
	reg.set("/npm/left-pad", npmDoc("1.0.0"))
	h, db := changeFeedRouter(t, reg)
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	c := changes.WithClient(client.NewClient(client.WithHTTPClient(reg.Client()), client.WithMaxRetries(0)))
	feeds := []changes.Feed{
		changes.NewPackagist(c, changes.WithURL(changes.PackagistFeedURL(reg.URL))),
		changes.NewNPM(c, changes.WithURL(reg.URL+"/replicate/registry/_changes")),
	}
	cursors := make([]changes.Cursor, len(feeds))
	for i, feed := range feeds {
		cursor, err := readChangeFeed(context.Background(), db, logger, feed, "")
		if err != nil || cursor == "" {
			t.Fatalf("starting %s feed: cursor %q, %v", feed.Ecosystem(), cursor, err)
		}
		cursors[i] = cursor
	}

	getBody(t, h, "/composer/p2/acme/lib.json")
	getBody(t, h, "/composer/p2/acme/other.json")
	getBody(t, h, "/npm/left-pad")

	reg.set("/p2/acme/lib.json", composerDoc("1.0.0", "1.1.0"))
	reg.set("/npm/left-pad", npmDoc("1.0.0", "1.1.0"))
	if body := getBody(t, h, "/composer/p2/acme/lib.json"); strings.Contains(body, "1.1.0") {
		t.Fatal("composer metadata revalidated before the feed listed it; the test proves nothing")
	}
	if body := getBody(t, h, "/npm/left-pad"); strings.Contains(body, "1.1.0") {
		t.Fatal("npm packument revalidated before the feed listed it; the test proves nothing")
	}

	for i, feed := range feeds {
		next, err := readChangeFeed(context.Background(), db, logger, feed, cursors[i])
		if err != nil || next == cursors[i] {
			t.Fatalf("reading %s feed: cursor %q -> %q, %v", feed.Ecosystem(), cursors[i], next, err)
		}
	}

	if body := getBody(t, h, "/composer/p2/acme/lib.json"); !strings.Contains(body, "1.1.0") {
		t.Errorf("composer metadata after the feed listed it = %s, want 1.1.0", body)
	}
	if body := getBody(t, h, "/npm/left-pad"); !strings.Contains(body, "1.1.0") {
		t.Errorf("npm packument after the feed listed it = %s, want 1.1.0", body)
	}
	getBody(t, h, "/composer/p2/acme/other.json")
	if n := reg.fetched("/p2/acme/other.json"); n != 1 {
		t.Errorf("unlisted package fetched %d times, want once", n)
	}
	if n := reg.fetched("/p2/acme/lib.json"); n != 2 {
		t.Errorf("listed package fetched %d times, want twice", n)
	}
}

// countingFeed fails its first reads and then reports no changes.
type countingFeed struct {
	reads    atomic.Int32
	failures int32
}

func (f *countingFeed) Ecosystem() string { return "npm" }

func (f *countingFeed) Start(context.Context) (changes.Cursor, error) {
	return f.read()
}

func (f *countingFeed) Changes(_ context.Context, cursor changes.Cursor) (changes.Changes, changes.Cursor, error) {
	if _, err := f.read(); err != nil {
		return changes.Changes{}, cursor, err
	}
	return changes.Changes{}, cursor, nil
}

func (f *countingFeed) read() (changes.Cursor, error) {
	if f.reads.Add(1) <= f.failures {
		return "", errors.New("feed down")
	}
	return "1", nil
}

// fixedFeed reports the same names on every read.
type fixedFeed struct{ names []string }

func (fixedFeed) Ecosystem() string                             { return "npm" }
func (fixedFeed) Start(context.Context) (changes.Cursor, error) { return "1", nil }
func (f fixedFeed) Changes(context.Context, changes.Cursor) (changes.Changes, changes.Cursor, error) {
	return changes.Changes{Names: f.names}, "2", nil
}

// TestReadChangeFeedKeepsCursorWhenExpiryFails checks that the cursor moves
// past the changes only once they are expired, so changes whose expiry
// failed are read again.
func TestReadChangeFeedKeepsCursorWhenExpiryFails(t *testing.T) {
	db, err := database.Create(filepath.Join(t.TempDir(), "proxy.db"))
	if err != nil {
		t.Fatal(err)
	}
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	feed := fixedFeed{names: []string{"left-pad"}}

	if next, err := readChangeFeed(context.Background(), db, logger, feed, "1"); err != nil || next != "2" {
		t.Fatalf("read = %q, %v; want cursor 2", next, err)
	}
	_ = db.Close()
	if next, err := readChangeFeed(context.Background(), db, logger, feed, "1"); err == nil || next != "1" {
		t.Errorf("read with a failing expiry = %q, %v; want an error and cursor 1", next, err)
	}
}

// TestFollowChangeFeedStopsWithContext checks that a failing feed is retried
// rather than given up on, and that following stops when ctx ends.
func TestFollowChangeFeedStopsWithContext(t *testing.T) {
	feed := &countingFeed{failures: 2}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		followChangeFeed(ctx, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), feed, time.Millisecond)
	}()
	deadline := time.After(5 * time.Second)
	for feed.reads.Load() < 4 {
		select {
		case <-deadline:
			t.Fatalf("feed read %d times, want it read again after failing", feed.reads.Load())
		case <-time.After(time.Millisecond):
		}
	}
	cancel()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("followChangeFeed did not return after ctx ended")
	}
}