Add gems virtual repositories

Merge the compact index across gems members and redirect downloads to the owning member.
2026-10-18 00:46:03 +11:00
parent 38641983e7
commit d035c6cec2
10 changed files with 633 additions and 2 deletions
@@ -32,6 +32,12 @@ type Engine struct {
 	// to the repo, or ErrNotFound.
 	NPMLocal func(ctx context.Context, repo, pkg string) ([]byte, error)

+	gemsMember func(ctx context.Context, name, path string) (*GemsMember, error)
+
+	// GemsLocal returns a local gems repo's versions, names or info/<gem>, or
+	// ErrNotFound.
+	GemsLocal func(ctx context.Context, repo, path string) ([]byte, error)
+
 	mergeTTL time.Duration

 	sf   singleflight.Group
@@ -61,6 +67,7 @@ func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
 	e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL}
 	e.rpmMember = e.fetchRPMMember
 	e.npmMember = e.fetchNPMMember
+	e.gemsMember = e.fetchGemsMember
 	return e
 }

@@ -202,6 +209,11 @@ func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path,
 	if !found || !slices.Contains(virt.Members, name) {
 		return "", false, nil
 	}
+	return e.memberLocation(ctx, name, rest, rawQuery, proxyBaseURL)
+}
+
+// memberLocation returns the URL of rest on member name's own route.
+func (e *Engine) memberLocation(ctx context.Context, name, rest, rawQuery, proxyBaseURL string) (string, bool, error) {
 	segs, err := memberSegments(rest)
 	if err != nil {
 		return "", false, err
@@ -0,0 +1,294 @@
+package virtual
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"slices"
+	"strings"
+	"time"
+
+	"git.unkin.net/unkin/artifactapi/internal/provider/gems"
+	"git.unkin.net/unkin/artifactapi/internal/proxy"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+const (
+	maxGemsInfos      = 1024
+	maxGemsCandidates = 8
+)
+
+// GemsMember is one member's compact index document; Body is nil when the
+// member does not have it.
+type GemsMember struct {
+	RemoteName string
+	Body       []byte
+}
+
+// GemsDoc is a merged compact index document. Owners maps each /info
+// version[-platform] to the member serving it.
+type GemsDoc struct {
+	Body   []byte
+	Owners map[string]string
+}
+
+func (d *GemsDoc) index() []byte            { return d.Body }
+func (d *GemsDoc) files() map[string][]byte { return nil }
+
+// MergeGemsInfo unions /info lines across members; the first member wins a
+// duplicate version-platform. A gem only one member has is passed through
+// verbatim so its md5 matches that member's /versions. ErrNotFound means no
+// member has the gem.
+func MergeGemsInfo(members []GemsMember) (*GemsDoc, error) {
+	var have []GemsMember
+	for _, m := range members {
+		if m.Body != nil {
+			have = append(have, m)
+		}
+	}
+	if len(have) == 0 {
+		return nil, ErrNotFound
+	}
+	d := &GemsDoc{Owners: map[string]string{}}
+	var lines []string
+	for _, m := range have {
+		for _, l := range gems.InfoLines(m.Body) {
+			k := gems.InfoKey(l)
+			if _, dup := d.Owners[k]; dup {
+				continue
+			}
+			d.Owners[k] = m.RemoteName
+			lines = append(lines, l)
+		}
+	}
+	if len(have) == 1 {
+		d.Body = have[0].Body
+	} else {
+		d.Body = gems.InfoBody(lines)
+	}
+	return d, nil
+}
+
+// versionsLines returns the entry lines of a /versions document.
+func versionsLines(m GemsMember) ([]byte, error) {
+	if m.Body == nil {
+		return nil, fmt.Errorf("member %q has no versions", m.RemoteName)
+	}
+	if bytes.HasPrefix(m.Body, []byte("---\n")) {
+		return m.Body[4:], nil
+	}
+	_, rest, ok := bytes.Cut(m.Body, []byte("\n---\n"))
+	if !ok {
+		return nil, fmt.Errorf("member %q versions has no --- separator", m.RemoteName)
+	}
+	return rest, nil
+}
+
+func gemName(line []byte) string {
+	n, _, _ := bytes.Cut(line, []byte(" "))
+	return string(bytes.TrimSpace(n))
+}
+
+// MergeGemsVersions passes through the lines of gems only one member lists
+// and writes one regenerated line, from the merged /info, per gem that
+// several members list.
+func MergeGemsVersions(members []GemsMember, info func(name string) (*GemsDoc, error), at time.Time) ([]byte, error) {
+	bodies := make([][]byte, len(members))
+	owners := map[string]int{}
+	for i, m := range members {
+		rest, err := versionsLines(m)
+		if err != nil {
+			return nil, err
+		}
+		bodies[i] = rest
+		seen := map[string]bool{}
+		for line := range bytes.Lines(rest) {
+			if n := gemName(line); n != "" && !seen[n] {
+				seen[n] = true
+				owners[n]++
+			}
+		}
+	}
+	var b bytes.Buffer
+	b.WriteString(gems.VersionsHeader(at))
+	var shared []string
+	for _, rest := range bodies {
+		for line := range bytes.Lines(rest) {
+			n := gemName(line)
+			if n == "" {
+				continue
+			}
+			if owners[n] > 1 {
+				if !slices.Contains(shared, n) {
+					shared = append(shared, n)
+				}
+				continue
+			}
+			b.Write(bytes.TrimRight(line, "\n"))
+			b.WriteByte('\n')
+		}
+	}
+	for _, n := range shared {
+		d, err := info(n)
+		if errors.Is(err, ErrNotFound) {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		b.WriteString(gems.VersionsLine(n, d.Body) + "\n")
+	}
+	return b.Bytes(), nil
+}
+
+// MergeGemsNames unions member /names documents, sorted.
+func MergeGemsNames(members []GemsMember) ([]byte, error) {
+	set := map[string]bool{}
+	for _, m := range members {
+		rest, err := versionsLines(m)
+		if err != nil {
+			return nil, err
+		}
+		for line := range bytes.Lines(rest) {
+			if n := gemName(line); n != "" {
+				set[n] = true
+			}
+		}
+	}
+	names := make([]string, 0, len(set))
+	for n := range set {
+		names = append(names, n)
+	}
+	slices.Sort(names)
+	return []byte("---\n" + strings.Join(names, "\n") + "\n"), nil
+}
+
+// FetchGems serves a gems virtual's merged versions, names or info/<gem>.
+func (e *Engine) FetchGems(ctx context.Context, virt models.Virtual, path string) ([]byte, string, error) {
+	var d *GemsDoc
+	var err error
+	switch {
+	case strings.HasPrefix(path, "info/"):
+		name := strings.TrimPrefix(path, "info/")
+		if !gems.NameRe.MatchString(name) {
+			return nil, "", ErrNotFound
+		}
+		d, err = e.gemsInfo(ctx, virt, name)
+	case path == "versions":
+		d, err = e.gemsMerged(ctx, virt, path, func(ctx context.Context, ms []GemsMember) ([]byte, error) {
+			return MergeGemsVersions(ms, func(name string) (*GemsDoc, error) { return e.gemsInfo(ctx, virt, name) }, time.Now())
+		})
+	case path == "names":
+		d, err = e.gemsMerged(ctx, virt, path, func(_ context.Context, ms []GemsMember) ([]byte, error) {
+			return MergeGemsNames(ms)
+		})
+	default:
+		return nil, "", ErrNotFound
+	}
+	if err != nil {
+		return nil, "", err
+	}
+	return d.Body, "text/plain; charset=utf-8", nil
+}
+
+func (e *Engine) gemsMerged(ctx context.Context, virt models.Virtual, path string, merge func(context.Context, []GemsMember) ([]byte, error)) (*GemsDoc, error) {
+	repo, err := e.merged(ctx, "gems\x00"+virt.Name+"\x00"+path, func(ctx context.Context) (mergedRepo, error) {
+		members, err := fetchMembers(virt, func(name string) (*GemsMember, error) { return e.gemsMember(ctx, name, path) })
+		if err != nil {
+			return nil, err
+		}
+		body, err := merge(ctx, members)
+		if err != nil {
+			return nil, err
+		}
+		return &GemsDoc{Body: body}, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return repo.(*GemsDoc), nil
+}
+
+func (e *Engine) gemsInfo(ctx context.Context, virt models.Virtual, name string) (*GemsDoc, error) {
+	prefix := "gems\x00" + virt.Name + "\x00info/"
+	inserted := false
+	repo, err := e.merged(ctx, prefix+name, func(ctx context.Context) (mergedRepo, error) {
+		inserted = true
+		members, err := fetchMembers(virt, func(m string) (*GemsMember, error) { return e.gemsMember(ctx, m, "info/"+name) })
+		if err != nil {
+			return nil, err
+		}
+		return MergeGemsInfo(members)
+	})
+	if err != nil {
+		return nil, err
+	}
+	if inserted {
+		e.evictOldest(prefix, maxGemsInfos)
+	}
+	return repo.(*GemsDoc), nil
+}
+
+// GemsRedirect maps a gems virtual download (gems/<full>.gem or
+// quick/Marshal.4.8/<full>.gemspec.rz) to the member whose /info line won the
+// merge for that version-platform.
+func (e *Engine) GemsRedirect(ctx context.Context, virt models.Virtual, path, rawQuery, proxyBaseURL string) (string, error) {
+	full, ok := strings.CutPrefix(path, "gems/")
+	if ok {
+		full, ok = strings.CutSuffix(full, ".gem")
+	} else if full, ok = strings.CutPrefix(path, "quick/Marshal.4.8/"); ok {
+		full, ok = strings.CutSuffix(full, ".gemspec.rz")
+	}
+	if !ok || strings.Contains(full, "/") {
+		return "", ErrNotFound
+	}
+	tried := 0
+	for i := 1; i < len(full)-1 && tried < maxGemsCandidates; i++ {
+		if full[i] != '-' || full[i+1] < '0' || full[i+1] > '9' || !gems.NameRe.MatchString(full[:i]) {
+			continue
+		}
+		tried++
+		d, err := e.gemsInfo(ctx, virt, full[:i])
+		if errors.Is(err, ErrNotFound) {
+			continue
+		}
+		if err != nil {
+			return "", err
+		}
+		if owner, ok := d.Owners[full[i+1:]]; ok {
+			loc, ok, err := e.memberLocation(ctx, owner, path, rawQuery, proxyBaseURL)
+			if err == nil && !ok {
+				err = ErrNotFound
+			}
+			return loc, err
+		}
+	}
+	return "", ErrNotFound
+}
+
+// fetchGemsMember returns a nil Body when the member lacks the document.
+func (e *Engine) fetchGemsMember(ctx context.Context, name, path string) (*GemsMember, error) {
+	remote, err := e.getRemote(ctx, name)
+	if err != nil {
+		return nil, fmt.Errorf("remote %q: %w", name, err)
+	}
+	m := &GemsMember{RemoteName: name}
+	if remote.RepoType == models.RepoTypeLocal {
+		if e.GemsLocal == nil {
+			return nil, errors.New("local gems indexes are not wired")
+		}
+		m.Body, err = e.GemsLocal(ctx, name, path)
+	} else {
+		m.Body, err = e.fetchMemberPath(ctx, *remote, path)
+	}
+	var pe *proxy.ProxyError
+	if errors.Is(err, ErrNotFound) || errors.As(err, &pe) && pe.Status == http.StatusNotFound {
+		return &GemsMember{RemoteName: name}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return m, nil
+}
@@ -0,0 +1,180 @@
+package virtual
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.unkin.net/unkin/artifactapi/internal/provider/gems"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+func TestMergeGemsInfoMemberOrderWins(t *testing.T) {
+	local := GemsMember{RemoteName: "internal", Body: []byte("---\n1.0.0 |checksum:aa\n1.0.0-java |checksum:ab\n")}
+	remote := GemsMember{RemoteName: "rubygems", Body: []byte("---\n0.9.0 |checksum:99\n1.0.0 |checksum:bb\n2.0.0 rack:>= 0|checksum:cc,ruby:>= 3.0\n")}
+	d, err := MergeGemsInfo([]GemsMember{local, {RemoteName: "empty"}, remote})
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "---\n1.0.0 |checksum:aa\n1.0.0-java |checksum:ab\n0.9.0 |checksum:99\n2.0.0 rack:>= 0|checksum:cc,ruby:>= 3.0\n"
+	if string(d.Body) != want {
+		t.Fatalf("merged info =\n%q\nwant\n%q", d.Body, want)
+	}
+	for k, owner := range map[string]string{"1.0.0": "internal", "1.0.0-java": "internal", "0.9.0": "rubygems", "2.0.0": "rubygems"} {
+		if d.Owners[k] != owner {
+			t.Errorf("owner of %s = %q, want %q", k, d.Owners[k], owner)
+		}
+	}
+
+	odd := GemsMember{RemoteName: "rubygems", Body: []byte("---\n1.0.0 |checksum:bb\n\n")}
+	if d, _ := MergeGemsInfo([]GemsMember{{RemoteName: "internal"}, odd}); string(d.Body) != string(odd.Body) {
+		t.Fatalf("a gem only one member has must pass through verbatim, got %q", d.Body)
+	}
+	if _, err := MergeGemsInfo([]GemsMember{{RemoteName: "a"}}); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("want ErrNotFound, got %v", err)
+	}
+}
+
+func TestMergeGemsVersions(t *testing.T) {
+	local := GemsMember{RemoteName: "internal", Body: []byte("created_at: 2026-10-18T00:00:00Z\n---\nhello 0.1.0 11\nrack 9.0.0 22\n")}
+	remote := GemsMember{RemoteName: "rubygems", Body: []byte("created_at: 2024-04-01T00:00:05Z\n---\n" +
+		"rack 1.0.0,2.0.0 33\nrails 7.0.0 44\nrack -1.0.0,3.0.0 55\n")}
+	merged := &GemsDoc{Body: []byte("---\n9.0.0 |checksum:x\n2.0.0 |checksum:y\n3.0.0 |checksum:z\n")}
+	var asked []string
+	at := time.Date(2026, 10, 18, 1, 2, 3, 0, time.UTC)
+	body, err := MergeGemsVersions([]GemsMember{local, remote}, func(name string) (*GemsDoc, error) {
+		asked = append(asked, name)
+		return merged, nil
+	}, at)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "created_at: 2026-10-18T01:02:03Z\n---\nhello 0.1.0 11\nrails 7.0.0 44\nrack 9.0.0,2.0.0,3.0.0 " + gems.MD5Hex(merged.Body) + "\n"
+	if string(body) != want {
+		t.Fatalf("versions =\n%q\nwant\n%q", body, want)
+	}
+	if len(asked) != 1 || asked[0] != "rack" {
+		t.Fatalf("merged info fetched for %v, want only the shared gem", asked)
+	}
+
+	if _, err := MergeGemsVersions([]GemsMember{{RemoteName: "x", Body: []byte("no separator")}}, nil, at); err == nil {
+		t.Fatal("malformed member versions must fail")
+	}
+	if _, err := MergeGemsVersions([]GemsMember{{RemoteName: "x"}}, nil, at); err == nil {
+		t.Fatal("a member without versions must fail")
+	}
+}
+
+func TestMergeGemsNames(t *testing.T) {
+	body, err := MergeGemsNames([]GemsMember{{RemoteName: "a", Body: []byte("---\nzeta\nalpha\n")}, {RemoteName: "b", Body: []byte("---\nalpha\nmid\n")}})
+	if err != nil || string(body) != "---\nalpha\nmid\nzeta\n" {
+		t.Fatalf("names = %q %v", body, err)
+	}
+}
+
+func gemsEngine(fetch func(name, path string) (*GemsMember, error)) *Engine {
+	return &Engine{
+		mergeTTL: time.Minute,
+		getRemote: func(_ context.Context, name string) (*models.Remote, error) {
+			rt := models.RepoTypeRemote
+			if name == "internal" {
+				rt = models.RepoTypeLocal
+			}
+			return &models.Remote{Name: name, RepoType: rt}, nil
+		},
+		gemsMember: func(_ context.Context, name, path string) (*GemsMember, error) { return fetch(name, path) },
+	}
+}
+
+func gemsVirt(members ...string) models.Virtual {
+	return models.Virtual{Name: "gems", PackageType: models.PackageGems, Members: members}
+}
+
+func TestGemsRedirectToOwner(t *testing.T) {
+	var mu sync.Mutex
+	asked := map[string]int{}
+	e := gemsEngine(func(name, path string) (*GemsMember, error) {
+		mu.Lock()
+		asked[name+" "+path]++
+		mu.Unlock()
+		switch name + " " + path {
+		case "internal info/foo-2fa":
+			return &GemsMember{RemoteName: name, Body: []byte("---\n1.0.0 |checksum:a\n")}, nil
+		case "rubygems info/foo-2fa":
+			return &GemsMember{RemoteName: name, Body: []byte("---\n1.0.0 |checksum:b\n1.1.0-x86_64-linux |checksum:c\n")}, nil
+		case "rubygems info/foo":
+			return &GemsMember{RemoteName: name, Body: []byte("---\n2.0.0 |checksum:d\n")}, nil
+		}
+		return &GemsMember{RemoteName: name}, nil
+	})
+	virt := gemsVirt("internal", "rubygems")
+	for path, want := range map[string]string{
+		"gems/foo-2fa-1.0.0.gem":                     "http://host/api/v1/local/internal/gems/foo-2fa-1.0.0.gem",
+		"gems/foo-2fa-1.1.0-x86_64-linux.gem":        "http://host/api/v1/remote/rubygems/gems/foo-2fa-1.1.0-x86_64-linux.gem",
+		"quick/Marshal.4.8/foo-2fa-1.0.0.gemspec.rz": "http://host/api/v1/local/internal/quick/Marshal.4.8/foo-2fa-1.0.0.gemspec.rz",
+		"quick/Marshal.4.8/foo-2.0.0.gemspec.rz":     "http://host/api/v1/remote/rubygems/quick/Marshal.4.8/foo-2.0.0.gemspec.rz",
+	} {
+		loc, err := e.GemsRedirect(context.Background(), virt, path, "", "http://host")
+		if err != nil || loc != want {
+			t.Errorf("%s -> %q %v, want %q", path, loc, err, want)
+		}
+	}
+	for _, path := range []string{"gems/foo-2fa-9.9.9.gem", "gems/nodash.gem", "gems/x/../foo-2.0.0.gem", "other/foo-2.0.0.gem", "specs.4.8.gz"} {
+		if _, err := e.GemsRedirect(context.Background(), virt, path, "", "http://host"); !errors.Is(err, ErrNotFound) {
+			t.Errorf("%s: want ErrNotFound, got %v", path, err)
+		}
+	}
+	if asked["internal info/foo-2fa"] != 1 {
+		t.Fatalf("merged info must be cached across redirects, internal asked %d times", asked["internal info/foo-2fa"])
+	}
+}
+
+func TestFetchGemsFailsClosed(t *testing.T) {
+	e := gemsEngine(func(name, path string) (*GemsMember, error) {
+		if name == "rubygems" {
+			return nil, errors.New("upstream down")
+		}
+		return &GemsMember{RemoteName: name, Body: []byte("created_at: x\n---\nhello 0.1.0 11\n")}, nil
+	})
+	for _, path := range []string{"versions", "names", "info/hello"} {
+		if _, _, err := e.FetchGems(context.Background(), gemsVirt("internal", "rubygems"), path); err == nil || errors.Is(err, ErrNotFound) {
+			t.Errorf("%s: want member error, got %v", path, err)
+		}
+	}
+	if _, err := e.GemsRedirect(context.Background(), gemsVirt("internal", "rubygems"), "gems/hello-0.1.0.gem", "", "http://h"); err == nil || errors.Is(err, ErrNotFound) {
+		t.Errorf("redirect: want member error, got %v", err)
+	}
+}
+
+func TestFetchGemsVersionsUsesMergedInfo(t *testing.T) {
+	e := gemsEngine(func(name, path string) (*GemsMember, error) {
+		docs := map[string]string{
+			"internal versions":  "created_at: x\n---\nrack 9.0.0 11\n",
+			"rubygems versions":  "created_at: y\n---\nrack 3.0.0 22\nrails 7.0.0 33\n",
+			"internal info/rack": "---\n9.0.0 |checksum:a\n",
+			"rubygems info/rack": "---\n3.0.0 |checksum:b\n",
+		}
+		if d, ok := docs[name+" "+path]; ok {
+			return &GemsMember{RemoteName: name, Body: []byte(d)}, nil
+		}
+		return &GemsMember{RemoteName: name}, nil
+	})
+	virt := gemsVirt("internal", "rubygems")
+	info, ct, err := e.FetchGems(context.Background(), virt, "info/rack")
+	if err != nil || !strings.HasPrefix(ct, "text/plain") {
+		t.Fatalf("info: %v %q", err, ct)
+	}
+	versions, _, err := e.FetchGems(context.Background(), virt, "versions")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(versions), "\nrails 7.0.0 33\nrack 9.0.0,3.0.0 "+gems.MD5Hex(info)+"\n") {
+		t.Fatalf("versions = %q", versions)
+	}
+	if _, _, err := e.FetchGems(context.Background(), virt, "info/../x"); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("bad name: %v", err)
+	}
+}